EG/plugins/user/navigation_mesh/utils/navmesh_serializer.py
2025-10-30 11:46:41 +08:00

728 lines
26 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 json
import zlib
import base64
from typing import Dict, Any, List, Optional
from panda3d.core import Point3
import time
class NavMeshSerializer:
"""
导航网格序列化器
处理导航网格的导入导出和版本兼容性
"""
def __init__(self):
"""初始化序列化器"""
self.current_version = "2.0"
self.supported_versions = ["1.0", "1.1", "2.0"]
def serialize_navmesh(self, navmesh_manager) -> str:
"""
序列化导航网格为字符串
Args:
navmesh_manager: 导航网格管理器
Returns:
序列化后的字符串
"""
try:
# 构建导航网格数据
data = self._build_navmesh_data(navmesh_manager)
# 转换为JSON字符串
json_str = json.dumps(data, ensure_ascii=False, separators=(',', ':'))
return json_str
except Exception as e:
print(f"序列化导航网格失败: {e}")
return ""
def deserialize_navmesh(self, data_str: str, navmesh_manager) -> bool:
"""
从字符串反序列化导航网格
Args:
data_str: 序列化数据字符串
navmesh_manager: 导航网格管理器
Returns:
是否反序列化成功
"""
try:
# 解析JSON数据
data = json.loads(data_str)
# 处理版本兼容性
data = self._handle_version_compatibility(data)
# 应用数据到导航网格管理器
self._apply_navmesh_data(data, navmesh_manager)
return True
except Exception as e:
print(f"反序列化导航网格失败: {e}")
return False
def save_navmesh_to_file(self, navmesh_manager, filepath: str,
compress: bool = False, pretty_print: bool = False) -> bool:
"""
保存导航网格到文件
Args:
navmesh_manager: 导航网格管理器
filepath: 文件路径
compress: 是否压缩
pretty_print: 是否美化输出
Returns:
是否保存成功
"""
try:
# 构建导航网格数据
data = self._build_navmesh_data(navmesh_manager)
# 添加保存时间戳
data['metadata']['saved_at'] = time.strftime("%Y-%m-%d %H:%M:%S")
# 转换为JSON字符串
if pretty_print:
json_str = json.dumps(data, ensure_ascii=False, indent=2)
else:
json_str = json.dumps(data, ensure_ascii=False, separators=(',', ':'))
# 处理压缩
if compress:
# 压缩数据
compressed_data = zlib.compress(json_str.encode('utf-8'))
encoded_data = base64.b64encode(compressed_data).decode('ascii')
# 保存压缩数据
save_data = {
'version': self.current_version,
'compressed': True,
'data': encoded_data
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(save_data, f, ensure_ascii=False, indent=2)
else:
# 直接保存JSON数据
with open(filepath, 'w', encoding='utf-8') as f:
f.write(json_str)
return True
except Exception as e:
print(f"保存导航网格失败: {e}")
return False
def load_navmesh_from_file(self, filepath: str, navmesh_manager) -> bool:
"""
从文件加载导航网格
Args:
filepath: 文件路径
navmesh_manager: 导航网格管理器
Returns:
是否加载成功
"""
try:
# 读取文件内容
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
# 尝试解析为JSON
try:
# 首先尝试解析为包装格式(可能包含压缩信息)
wrapper_data = json.loads(content)
if isinstance(wrapper_data, dict) and 'compressed' in wrapper_data:
if wrapper_data.get('compressed', False):
# 解压缩数据
encoded_data = wrapper_data['data']
compressed_data = base64.b64decode(encoded_data.encode('ascii'))
json_str = zlib.decompress(compressed_data).decode('utf-8')
data = json.loads(json_str)
else:
data = wrapper_data
else:
# 直接使用内容
data = wrapper_data
except json.JSONDecodeError:
# 如果不是JSON格式假设是直接的导航网格数据
data = json.loads(content)
# 处理版本兼容性
data = self._handle_version_compatibility(data)
# 应用数据到导航网格管理器
self._apply_navmesh_data(data, navmesh_manager)
return True
except Exception as e:
print(f"加载导航网格失败: {e}")
return False
def _build_navmesh_data(self, navmesh_manager) -> Dict[str, Any]:
"""
构建导航网格数据
Args:
navmesh_manager: 导航网格管理器
Returns:
导航网格数据字典
"""
# 构建元数据
metadata = {
'version': self.current_version,
'generator': 'EG Navigation Mesh System',
'created': time.strftime("%Y-%m-%d %H:%M:%S"),
'polygon_count': len(navmesh_manager.polygons) if navmesh_manager.polygons else 0,
'obstacle_count': len(navmesh_manager.obstacles) if navmesh_manager.obstacles else 0
}
# 添加设置信息
settings = {
'cell_size': getattr(navmesh_manager, 'cell_size', 0.5),
'cell_height': getattr(navmesh_manager, 'cell_height', 0.2),
'agent_height': getattr(navmesh_manager, 'agent_height', 1.8),
'agent_radius': getattr(navmesh_manager, 'agent_radius', 0.3),
'max_slope': getattr(navmesh_manager, 'max_slope', 45.0),
'max_climb': getattr(navmesh_manager, 'max_climb', 0.2),
'min_region_area': getattr(navmesh_manager, 'min_region_area', 3),
'merge_region_area': getattr(navmesh_manager, 'merge_region_area', 20),
'edge_max_len': getattr(navmesh_manager, 'edge_max_len', 12.0),
'edge_max_error': getattr(navmesh_manager, 'edge_max_error', 1.3),
'verts_per_poly': getattr(navmesh_manager, 'verts_per_poly', 6)
}
# 构建多边形数据
polygons = []
if navmesh_manager.polygons:
for polygon_id, polygon in navmesh_manager.polygons.items():
poly_data = {
'id': polygon_id,
'vertices': [[v.x, v.y, v.z] for v in polygon.vertices],
'neighbors': polygon.neighbors,
'portal_points': {},
'flags': getattr(polygon, 'flags', 0),
'area': getattr(polygon, 'area', 0.0)
}
# 添加门户点数据
for neighbor_id, (left_point, right_point) in polygon.portal_points.items():
poly_data['portal_points'][str(neighbor_id)] = [
[left_point.x, left_point.y, left_point.z],
[right_point.x, right_point.y, right_point.z]
]
polygons.append(poly_data)
# 构建障碍物数据
obstacles = []
if navmesh_manager.obstacles:
for obstacle_id, obstacle in navmesh_manager.obstacles.items():
obs_data = {
'id': obstacle_id,
'position': [
obstacle['position'].x,
obstacle['position'].y,
obstacle['position'].z
],
'radius': obstacle['radius'],
'height': obstacle['height'],
'type': obstacle.get('type', 'cylinder')
}
obstacles.append(obs_data)
# 构建完整数据结构
data = {
'metadata': metadata,
'settings': settings,
'polygons': polygons,
'obstacles': obstacles
}
return data
def _handle_version_compatibility(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
处理版本兼容性
Args:
data: 导航网格数据
Returns:
处理后的数据
"""
if 'metadata' not in data or 'version' not in data['metadata']:
# 假设是旧版本数据
data = self._upgrade_from_version_1_0(data)
version = data['metadata']['version']
# 根据版本进行升级
if version == "1.0":
data = self._upgrade_from_version_1_0(data)
elif version == "1.1":
data = self._upgrade_from_version_1_1(data)
# 版本2.0不需要升级
return data
def _upgrade_from_version_1_0(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
从版本1.0升级
Args:
data: 版本1.0的数据
Returns:
升级后的数据
"""
# 添加缺失的字段
if 'metadata' not in data:
data['metadata'] = {}
data['metadata']['version'] = self.current_version
# 确保所有必需字段存在
if 'settings' not in data:
data['settings'] = {}
# 设置默认值
default_settings = {
'cell_size': 0.5,
'cell_height': 0.2,
'agent_height': 1.8,
'agent_radius': 0.3,
'max_slope': 45.0,
'max_climb': 0.2,
'min_region_area': 3,
'merge_region_area': 20,
'edge_max_len': 12.0,
'edge_max_error': 1.3,
'verts_per_poly': 6
}
for key, value in default_settings.items():
if key not in data['settings']:
data['settings'][key] = value
return data
def _upgrade_from_version_1_1(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
从版本1.1升级
Args:
data: 版本1.1的数据
Returns:
升级后的数据
"""
# 版本1.1到2.0的升级
data['metadata']['version'] = self.current_version
return data
def _apply_navmesh_data(self, data: Dict[str, Any], navmesh_manager):
"""
应用导航网格数据到管理器
Args:
data: 导航网格数据
navmesh_manager: 导航网格管理器
"""
from ..core.navmesh_manager import NavMeshPolygon
# 清除现有数据
navmesh_manager.polygons.clear()
navmesh_manager.obstacles.clear()
# 应用设置
if 'settings' in data:
settings = data['settings']
for key, value in settings.items():
if hasattr(navmesh_manager, key):
setattr(navmesh_manager, key, value)
# 解析多边形
if 'polygons' in data:
for poly_data in data['polygons']:
polygon_id = poly_data['id']
vertices = [Point3(v[0], v[1], v[2]) for v in poly_data['vertices']]
polygon = NavMeshPolygon(polygon_id, vertices)
# 添加邻居信息
if 'neighbors' in poly_data:
for neighbor_id in poly_data['neighbors']:
polygon.add_neighbor(neighbor_id)
# 添加门户点信息
if 'portal_points' in poly_data:
for neighbor_id, portal_data in poly_data['portal_points'].items():
left_point = Point3(portal_data[0][0], portal_data[0][1], portal_data[0][2])
right_point = Point3(portal_data[1][0], portal_data[1][1], portal_data[1][2])
polygon.set_portal_points(int(neighbor_id), left_point, right_point)
# 设置标志位
if 'flags' in poly_data:
polygon.flags = poly_data['flags']
# 设置面积
if 'area' in poly_data:
polygon.area = poly_data['area']
navmesh_manager.polygons[polygon_id] = polygon
# 解析障碍物
if 'obstacles' in data:
for obs_data in data['obstacles']:
obstacle_id = obs_data['id']
navmesh_manager.obstacles[obstacle_id] = {
'position': Point3(obs_data['position'][0], obs_data['position'][1], obs_data['position'][2]),
'radius': obs_data['radius'],
'height': obs_data['height'],
'type': obs_data.get('type', 'cylinder')
}
# 更新统计信息
navmesh_manager._update_stats()
# 更新空间哈希
navmesh_manager._update_spatial_hash()
def export_to_standard_format(self, navmesh_manager, filepath: str,
format_type: str = 'obj') -> bool:
"""
导出为标准格式
Args:
navmesh_manager: 导航网格管理器
filepath: 导出文件路径
format_type: 导出格式 ('obj', 'ply')
Returns:
是否导出成功
"""
try:
if format_type.lower() == 'obj':
return self._export_to_obj(navmesh_manager, filepath)
elif format_type.lower() == 'ply':
return self._export_to_ply(navmesh_manager, filepath)
else:
print(f"不支持的导出格式: {format_type}")
return False
except Exception as e:
print(f"导出为标准格式失败: {e}")
return False
def _export_to_obj(self, navmesh_manager, filepath: str) -> bool:
"""
导出为OBJ格式
Args:
navmesh_manager: 导航网格管理器
filepath: 导出文件路径
Returns:
是否导出成功
"""
try:
with open(filepath, 'w', encoding='utf-8') as f:
f.write("# Navigation Mesh Export\n")
f.write(f"# Generated by EG Navigation Mesh System\n")
f.write(f"# Polygons: {len(navmesh_manager.polygons)}\n\n")
# 写入顶点
vertex_index = 1
vertex_mapping = {} # 旧顶点索引到新顶点索引的映射
for polygon_id, polygon in navmesh_manager.polygons.items():
for vertex in polygon.vertices:
vertex_key = (vertex.x, vertex.y, vertex.z)
if vertex_key not in vertex_mapping:
f.write(f"v {vertex.x} {vertex.y} {vertex.z}\n")
vertex_mapping[vertex_key] = vertex_index
vertex_index += 1
f.write("\n")
# 写入面
for polygon_id, polygon in navmesh_manager.polygons.items():
f.write(f"g polygon_{polygon_id}\n")
face_vertices = []
for vertex in polygon.vertices:
vertex_key = (vertex.x, vertex.y, vertex.z)
face_vertices.append(str(vertex_mapping[vertex_key]))
f.write(f"f {' '.join(face_vertices)}\n\n")
return True
except Exception as e:
print(f"导出为OBJ格式失败: {e}")
return False
def _export_to_ply(self, navmesh_manager, filepath: str) -> bool:
"""
导出为PLY格式
Args:
navmesh_manager: 导航网格管理器
filepath: 导出文件路径
Returns:
是否导出成功
"""
try:
# 收集所有唯一顶点
vertices = []
vertex_mapping = {}
faces = []
for polygon_id, polygon in navmesh_manager.polygons.items():
face_indices = []
for vertex in polygon.vertices:
vertex_key = (vertex.x, vertex.y, vertex.z)
if vertex_key not in vertex_mapping:
vertex_mapping[vertex_key] = len(vertices)
vertices.append(vertex)
face_indices.append(vertex_mapping[vertex_key])
faces.append(face_indices)
# 写入PLY文件
with open(filepath, 'w', encoding='utf-8') as f:
# PLY头部
f.write("ply\n")
f.write("format ascii 1.0\n")
f.write(f"element vertex {len(vertices)}\n")
f.write("property float x\n")
f.write("property float y\n")
f.write("property float z\n")
f.write(f"element face {len(faces)}\n")
f.write("property list uchar int vertex_indices\n")
f.write("end_header\n")
# 写入顶点
for vertex in vertices:
f.write(f"{vertex.x} {vertex.y} {vertex.z}\n")
# 写入面
for face in faces:
f.write(f"{len(face)} {' '.join(map(str, face))}\n")
return True
except Exception as e:
print(f"导出为PLY格式失败: {e}")
return False
def import_from_standard_format(self, filepath: str, navmesh_manager,
format_type: str = 'obj') -> bool:
"""
从标准格式导入
Args:
filepath: 导入文件路径
navmesh_manager: 导航网格管理器
format_type: 导入格式 ('obj', 'ply')
Returns:
是否导入成功
"""
try:
if format_type.lower() == 'obj':
return self._import_from_obj(filepath, navmesh_manager)
elif format_type.lower() == 'ply':
return self._import_from_ply(filepath, navmesh_manager)
else:
print(f"不支持的导入格式: {format_type}")
return False
except Exception as e:
print(f"从标准格式导入失败: {e}")
return False
def _import_from_obj(self, filepath: str, navmesh_manager) -> bool:
"""
从OBJ格式导入
Args:
filepath: 导入文件路径
navmesh_manager: 导航网格管理器
Returns:
是否导入成功
"""
try:
vertices = []
faces = []
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
parts = line.split()
if not parts:
continue
if parts[0] == 'v':
# 顶点
if len(parts) >= 4:
x, y, z = float(parts[1]), float(parts[2]), float(parts[3])
vertices.append(Point3(x, y, z))
elif parts[0] == 'f':
# 面
face_indices = []
for part in parts[1:]:
# 处理索引OBJ格式索引从1开始
index = int(part.split('/')[0]) - 1
face_indices.append(index)
faces.append(face_indices)
# 创建多边形
navmesh_manager.polygons.clear()
for i, face_indices in enumerate(faces):
polygon_vertices = [vertices[idx] for idx in face_indices if 0 <= idx < len(vertices)]
if len(polygon_vertices) >= 3:
navmesh_manager.add_polygon(polygon_vertices)
return True
except Exception as e:
print(f"从OBJ格式导入失败: {e}")
return False
def _import_from_ply(self, filepath: str, navmesh_manager) -> bool:
"""
从PLY格式导入
Args:
filepath: 导入文件路径
navmesh_manager: 导航网格管理器
Returns:
是否导入成功
"""
try:
vertices = []
faces = []
with open(filepath, 'r', encoding='utf-8') as f:
lines = f.readlines()
# 解析PLY头部
i = 0
vertex_count = 0
face_count = 0
while i < len(lines):
line = lines[i].strip()
if not line:
i += 1
continue
if line == "end_header":
i += 1
break
parts = line.split()
if len(parts) >= 3 and parts[0] == "element":
if parts[1] == "vertex":
vertex_count = int(parts[2])
elif parts[1] == "face":
face_count = int(parts[2])
i += 1
# 读取顶点
for j in range(vertex_count):
if i + j < len(lines):
parts = lines[i + j].strip().split()
if len(parts) >= 3:
x, y, z = float(parts[0]), float(parts[1]), float(parts[2])
vertices.append(Point3(x, y, z))
i += vertex_count
# 读取面
for j in range(face_count):
if i + j < len(lines):
parts = lines[i + j].strip().split()
if len(parts) >= 2:
count = int(parts[0])
face_indices = [int(idx) for idx in parts[1:1+count]]
faces.append(face_indices)
# 创建多边形
navmesh_manager.polygons.clear()
for face_indices in faces:
polygon_vertices = [vertices[idx] for idx in face_indices if 0 <= idx < len(vertices)]
if len(polygon_vertices) >= 3:
navmesh_manager.add_polygon(polygon_vertices)
return True
except Exception as e:
print(f"从PLY格式导入失败: {e}")
return False
def get_file_info(self, filepath: str) -> Dict[str, Any]:
"""
获取导航网格文件信息
Args:
filepath: 文件路径
Returns:
文件信息字典
"""
try:
info = {
'filepath': filepath,
'file_size': 0,
'version': 'unknown',
'polygon_count': 0,
'obstacle_count': 0,
'compressed': False
}
import os
info['file_size'] = os.path.getsize(filepath)
# 读取文件内容
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read(1024) # 只读取前1024个字符
# 尝试解析
try:
data = json.loads(content)
if isinstance(data, dict):
if 'compressed' in data:
info['compressed'] = data['compressed']
info['version'] = data.get('version', 'unknown')
elif 'metadata' in data:
info['version'] = data['metadata'].get('version', 'unknown')
info['polygon_count'] = data['metadata'].get('polygon_count', 0)
info['obstacle_count'] = data['metadata'].get('obstacle_count', 0)
except json.JSONDecodeError:
pass
return info
except Exception as e:
print(f"获取文件信息失败: {e}")
return {}