EG/plugins/user/procedural_terrain_generation/serialization/terrain_serializer.py
2025-12-12 16:16:15 +08:00

1416 lines
53 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 numpy as np
import json
import pickle
import zlib
import base64
from typing import Dict, Any, Optional
import time
import os
class TerrainSerializer:
"""
地形序列化器
负责地形数据的序列化和反序列化,支持多种格式和压缩方式
"""
def __init__(self, plugin):
"""
初始化地形序列化器
Args:
plugin: 程序化地形生成插件实例
"""
self.plugin = plugin
self.enabled = False
self.initialized = False
# 序列化配置
self.serialization_params = {
'default_format': 'json',
'compression_enabled': True,
'compression_level': 6,
'chunk_size': 1024,
'include_metadata': True,
'include_textures': True,
'include_vegetation': True,
'include_water': True
}
# 支持的格式
self.supported_formats = ['json', 'pickle', 'binary', 'custom']
# 统计信息
self.stats = {
'terrains_saved': 0,
'terrains_loaded': 0,
'total_save_time': 0.0,
'total_load_time': 0.0,
'average_save_time': 0.0,
'average_load_time': 0.0,
'data_compressed': 0,
'data_uncompressed': 0
}
print("✓ 地形序列化器已创建")
def initialize(self) -> bool:
"""
初始化地形序列化器
Returns:
是否初始化成功
"""
try:
self.initialized = True
print("✓ 地形序列化器初始化完成")
return True
except Exception as e:
print(f"✗ 地形序列化器初始化失败: {e}")
import traceback
traceback.print_exc()
return False
def enable(self) -> bool:
"""
启用地形序列化器
Returns:
是否启用成功
"""
try:
if not self.initialized:
print("✗ 地形序列化器未初始化")
return False
self.enabled = True
print("✓ 地形序列化器已启用")
return True
except Exception as e:
print(f"✗ 地形序列化器启用失败: {e}")
import traceback
traceback.print_exc()
return False
def disable(self):
"""禁用地形序列化器"""
try:
self.enabled = False
print("✓ 地形序列化器已禁用")
except Exception as e:
print(f"✗ 地形序列化器禁用失败: {e}")
import traceback
traceback.print_exc()
def finalize(self):
"""清理地形序列化器资源"""
try:
self.disable()
self.initialized = False
print("✓ 地形序列化器资源已清理")
except Exception as e:
print(f"✗ 地形序列化器资源清理失败: {e}")
import traceback
traceback.print_exc()
def update(self, dt: float):
"""
更新地形序列化器状态
Args:
dt: 时间增量
"""
# 处理更新逻辑
pass
def save_terrain(self, filename: str, terrain_data: Dict[str, Any] = None,
format: str = None, compress: bool = None) -> bool:
"""
保存地形数据
Args:
filename: 文件名
terrain_data: 地形数据字典如果为None则从地形管理器获取
format: 保存格式
compress: 是否压缩
Returns:
是否保存成功
"""
try:
save_start_time = time.time()
if not self.enabled:
print("✗ 地形序列化器未启用")
return False
# 使用默认参数如果未指定
if format is None:
format = self.serialization_params['default_format']
if compress is None:
compress = self.serialization_params['compression_enabled']
# 验证格式
if format not in self.supported_formats:
print(f"✗ 不支持的格式: {format}")
return False
print(f"✓ 开始保存地形到 {filename} (格式: {format}, 压缩: {compress})...")
# 如果未提供地形数据,则从地形管理器获取
if terrain_data is None:
if self.plugin.terrain_manager:
terrain_data = self.plugin.terrain_manager.get_terrain_data()
else:
print("✗ 无法获取地形数据")
return False
# 准备保存数据
save_data = self._prepare_terrain_data(terrain_data)
# 根据格式保存
if format == 'json':
success = self._save_json(filename, save_data, compress)
elif format == 'pickle':
success = self._save_pickle(filename, save_data, compress)
elif format == 'binary':
success = self._save_binary(filename, save_data, compress)
elif format == 'custom':
success = self._save_custom(filename, save_data, compress)
else:
print(f"✗ 不支持的保存格式: {format}")
return False
if success:
# 更新统计信息
save_time = time.time() - save_start_time
self.stats['terrains_saved'] += 1
self.stats['total_save_time'] += save_time
if self.stats['terrains_saved'] > 0:
self.stats['average_save_time'] = (
self.stats['total_save_time'] / self.stats['terrains_saved']
)
print(f"✓ 地形保存完成,耗时: {save_time:.3f}")
return True
else:
print("✗ 地形保存失败")
return False
except Exception as e:
print(f"✗ 地形保存失败: {e}")
import traceback
traceback.print_exc()
return False
def load_terrain(self, filename: str, format: str = None) -> Optional[Dict[str, Any]]:
"""
加载地形数据
Args:
filename: 文件名
format: 加载格式如果为None则自动检测
Returns:
地形数据字典或None
"""
try:
load_start_time = time.time()
if not self.enabled:
print("✗ 地形序列化器未启用")
return None
# 检查文件是否存在
if not os.path.exists(filename):
print(f"✗ 文件不存在: {filename}")
return None
print(f"✓ 开始加载地形从 {filename}...")
# 如果未指定格式,则尝试自动检测
if format is None:
format = self._detect_format(filename)
# 验证格式
if format not in self.supported_formats:
print(f"✗ 不支持的格式: {format}")
return None
# 根据格式加载
if format == 'json':
terrain_data = self._load_json(filename)
elif format == 'pickle':
terrain_data = self._load_pickle(filename)
elif format == 'binary':
terrain_data = self._load_binary(filename)
elif format == 'custom':
terrain_data = self._load_custom(filename)
else:
print(f"✗ 不支持的加载格式: {format}")
return None
if terrain_data is not None:
# 处理加载的数据
processed_data = self._process_loaded_data(terrain_data)
# 更新统计信息
load_time = time.time() - load_start_time
self.stats['terrains_loaded'] += 1
self.stats['total_load_time'] += load_time
if self.stats['terrains_loaded'] > 0:
self.stats['average_load_time'] = (
self.stats['total_load_time'] / self.stats['terrains_loaded']
)
print(f"✓ 地形加载完成,耗时: {load_time:.3f}")
return processed_data
else:
print("✗ 地形加载失败")
return None
except Exception as e:
print(f"✗ 地形加载失败: {e}")
import traceback
traceback.print_exc()
return None
def _prepare_terrain_data(self, terrain_data: Dict[str, Any]) -> Dict[str, Any]:
"""
准备地形数据以供保存
Args:
terrain_data: 原始地形数据
Returns:
准备好的地形数据
"""
try:
# 创建保存数据的副本
save_data = {
'version': '1.0',
'timestamp': time.time(),
'format': self.serialization_params['default_format']
}
# 添加地形数据
if 'heightmap' in terrain_data and terrain_data['heightmap'] is not None:
heightmap = terrain_data['heightmap']
save_data['heightmap'] = {
'data': heightmap.tolist() if isinstance(heightmap, np.ndarray) else heightmap,
'shape': heightmap.shape if isinstance(heightmap, np.ndarray) else None,
'dtype': str(heightmap.dtype) if isinstance(heightmap, np.ndarray) else 'unknown'
}
# 添加法线图
if 'normalmap' in terrain_data and terrain_data['normalmap'] is not None:
normalmap = terrain_data['normalmap']
save_data['normalmap'] = {
'data': normalmap.tolist() if isinstance(normalmap, np.ndarray) else normalmap,
'shape': normalmap.shape if isinstance(normalmap, np.ndarray) else None,
'dtype': str(normalmap.dtype) if isinstance(normalmap, np.ndarray) else 'unknown'
}
# 添加生物群落图
if self.serialization_params['include_metadata']:
if 'biome_map' in terrain_data and terrain_data['biome_map'] is not None:
biome_map = terrain_data['biome_map']
save_data['biome_map'] = {
'data': biome_map.tolist() if isinstance(biome_map, np.ndarray) else biome_map,
'shape': biome_map.shape if isinstance(biome_map, np.ndarray) else None,
'dtype': str(biome_map.dtype) if isinstance(biome_map, np.ndarray) else 'unknown'
}
# 添加湿度图
if 'moisture_map' in terrain_data and terrain_data['moisture_map'] is not None:
moisture_map = terrain_data['moisture_map']
save_data['moisture_map'] = {
'data': moisture_map.tolist() if isinstance(moisture_map, np.ndarray) else moisture_map,
'shape': moisture_map.shape if isinstance(moisture_map, np.ndarray) else None,
'dtype': str(moisture_map.dtype) if isinstance(moisture_map, np.ndarray) else 'unknown'
}
# 添加温度图
if 'temperature_map' in terrain_data and terrain_data['temperature_map'] is not None:
temperature_map = terrain_data['temperature_map']
save_data['temperature_map'] = {
'data': temperature_map.tolist() if isinstance(temperature_map, np.ndarray) else temperature_map,
'shape': temperature_map.shape if isinstance(temperature_map, np.ndarray) else None,
'dtype': str(temperature_map.dtype) if isinstance(temperature_map, np.ndarray) else 'unknown'
}
# 添加侵蚀数据
if 'erosion_map' in terrain_data and terrain_data['erosion_map'] is not None:
erosion_map = terrain_data['erosion_map']
save_data['erosion_map'] = {
'data': erosion_map.tolist() if isinstance(erosion_map, np.ndarray) else erosion_map,
'shape': erosion_map.shape if isinstance(erosion_map, np.ndarray) else None,
'dtype': str(erosion_map.dtype) if isinstance(erosion_map, np.ndarray) else 'unknown'
}
# 添加沉积物数据
if 'sediment_map' in terrain_data and terrain_data['sediment_map'] is not None:
sediment_map = terrain_data['sediment_map']
save_data['sediment_map'] = {
'data': sediment_map.tolist() if isinstance(sediment_map, np.ndarray) else sediment_map,
'shape': sediment_map.shape if isinstance(sediment_map, np.ndarray) else None,
'dtype': str(sediment_map.dtype) if isinstance(sediment_map, np.ndarray) else 'unknown'
}
# 添加水图
if self.serialization_params['include_water']:
if 'water_map' in terrain_data and terrain_data['water_map'] is not None:
water_map = terrain_data['water_map']
save_data['water_map'] = {
'data': water_map.tolist() if isinstance(water_map, np.ndarray) else water_map,
'shape': water_map.shape if isinstance(water_map, np.ndarray) else None,
'dtype': str(water_map.dtype) if isinstance(water_map, np.ndarray) else 'unknown'
}
# 添加植被数据
if self.serialization_params['include_vegetation']:
if 'vegetation_map' in terrain_data and terrain_data['vegetation_map'] is not None:
vegetation_map = terrain_data['vegetation_map']
save_data['vegetation_map'] = {
'data': vegetation_map.tolist() if isinstance(vegetation_map, np.ndarray) else vegetation_map,
'shape': vegetation_map.shape if isinstance(vegetation_map, np.ndarray) else None,
'dtype': str(vegetation_map.dtype) if isinstance(vegetation_map, np.ndarray) else 'unknown'
}
# 添加统计信息
if 'stats' in terrain_data:
save_data['stats'] = terrain_data['stats']
return save_data
except Exception as e:
print(f"✗ 地形数据准备失败: {e}")
return {}
def _process_loaded_data(self, loaded_data: Dict[str, Any]) -> Dict[str, Any]:
"""
处理加载的地形数据
Args:
loaded_data: 加载的地形数据
Returns:
处理后的地形数据
"""
try:
# 创建处理后的数据
processed_data = {}
# 处理高度图
if 'heightmap' in loaded_data:
heightmap_data = loaded_data['heightmap']
if 'data' in heightmap_data:
processed_data['heightmap'] = np.array(heightmap_data['data'], dtype=np.float32)
# 处理法线图
if 'normalmap' in loaded_data:
normalmap_data = loaded_data['normalmap']
if 'data' in normalmap_data:
processed_data['normalmap'] = np.array(normalmap_data['data'], dtype=np.float32)
# 处理生物群落图
if 'biome_map' in loaded_data:
biome_data = loaded_data['biome_map']
if 'data' in biome_data:
processed_data['biome_map'] = np.array(biome_data['data'], dtype=np.int32)
# 处理湿度图
if 'moisture_map' in loaded_data:
moisture_data = loaded_data['moisture_map']
if 'data' in moisture_data:
processed_data['moisture_map'] = np.array(moisture_data['data'], dtype=np.float32)
# 处理温度图
if 'temperature_map' in loaded_data:
temperature_data = loaded_data['temperature_map']
if 'data' in temperature_data:
processed_data['temperature_map'] = np.array(temperature_data['data'], dtype=np.float32)
# 处理侵蚀图
if 'erosion_map' in loaded_data:
erosion_data = loaded_data['erosion_map']
if 'data' in erosion_data:
processed_data['erosion_map'] = np.array(erosion_data['data'], dtype=np.float32)
# 处理沉积物图
if 'sediment_map' in loaded_data:
sediment_data = loaded_data['sediment_map']
if 'data' in sediment_data:
processed_data['sediment_map'] = np.array(sediment_data['data'], dtype=np.float32)
# 处理水图
if 'water_map' in loaded_data:
water_data = loaded_data['water_map']
if 'data' in water_data:
processed_data['water_map'] = np.array(water_data['data'], dtype=np.float32)
# 处理植被图
if 'vegetation_map' in loaded_data:
vegetation_data = loaded_data['vegetation_map']
if 'data' in vegetation_data:
processed_data['vegetation_map'] = np.array(vegetation_data['data'], dtype=np.int32)
# 处理统计信息
if 'stats' in loaded_data:
processed_data['stats'] = loaded_data['stats']
return processed_data
except Exception as e:
print(f"✗ 加载数据处理失败: {e}")
return {}
def _save_json(self, filename: str, data: Dict[str, Any], compress: bool) -> bool:
"""
以JSON格式保存地形数据
Args:
filename: 文件名
data: 地形数据
compress: 是否压缩
Returns:
是否保存成功
"""
try:
# 序列化为JSON
json_data = json.dumps(data, ensure_ascii=False, indent=2)
if compress:
# 压缩数据
compressed_data = zlib.compress(json_data.encode('utf-8'),
self.serialization_params['compression_level'])
with open(filename, 'wb') as f:
f.write(compressed_data)
self.stats['data_compressed'] += len(compressed_data)
self.stats['data_uncompressed'] += len(json_data.encode('utf-8'))
else:
# 不压缩
with open(filename, 'w', encoding='utf-8') as f:
f.write(json_data)
self.stats['data_uncompressed'] += len(json_data.encode('utf-8'))
return True
except Exception as e:
print(f"✗ JSON格式地形保存失败: {e}")
return False
def _save_pickle(self, filename: str, data: Dict[str, Any], compress: bool) -> bool:
"""
以Pickle格式保存地形数据
Args:
filename: 文件名
data: 地形数据
compress: 是否压缩
Returns:
是否保存成功
"""
try:
# 序列化为Pickle
pickle_data = pickle.dumps(data)
if compress:
# 压缩数据
compressed_data = zlib.compress(pickle_data,
self.serialization_params['compression_level'])
with open(filename, 'wb') as f:
f.write(compressed_data)
self.stats['data_compressed'] += len(compressed_data)
self.stats['data_uncompressed'] += len(pickle_data)
else:
# 不压缩
with open(filename, 'wb') as f:
f.write(pickle_data)
self.stats['data_uncompressed'] += len(pickle_data)
return True
except Exception as e:
print(f"✗ Pickle格式地形保存失败: {e}")
return False
def _save_binary(self, filename: str, data: Dict[str, Any], compress: bool) -> bool:
"""
以二进制格式保存地形数据
Args:
filename: 文件名
data: 地形数据
compress: 是否压缩
Returns:
是否保存成功
"""
try:
# 转换为二进制格式
binary_data = self._convert_to_binary(data)
if compress:
# 压缩数据
compressed_data = zlib.compress(binary_data,
self.serialization_params['compression_level'])
with open(filename, 'wb') as f:
f.write(compressed_data)
self.stats['data_compressed'] += len(compressed_data)
self.stats['data_uncompressed'] += len(binary_data)
else:
# 不压缩
with open(filename, 'wb') as f:
f.write(binary_data)
self.stats['data_uncompressed'] += len(binary_data)
return True
except Exception as e:
print(f"✗ 二进制格式地形保存失败: {e}")
return False
def _save_custom(self, filename: str, data: Dict[str, Any], compress: bool) -> bool:
"""
以自定义格式保存地形数据
Args:
filename: 文件名
data: 地形数据
compress: 是否压缩
Returns:
是否保存成功
"""
try:
# 自定义格式保存这里简化为JSON格式
return self._save_json(filename, data, compress)
except Exception as e:
print(f"✗ 自定义格式地形保存失败: {e}")
return False
def _load_json(self, filename: str) -> Optional[Dict[str, Any]]:
"""
从JSON格式加载地形数据
Args:
filename: 文件名
Returns:
地形数据字典或None
"""
try:
# 读取文件
with open(filename, 'rb') as f:
file_data = f.read()
# 检查是否压缩
try:
# 尝试解压
decompressed_data = zlib.decompress(file_data)
json_data = decompressed_data.decode('utf-8')
self.stats['data_compressed'] += len(file_data)
self.stats['data_uncompressed'] += len(decompressed_data)
except zlib.error:
# 未压缩的JSON
json_data = file_data.decode('utf-8')
self.stats['data_uncompressed'] += len(file_data)
# 解析JSON
data = json.loads(json_data)
return data
except Exception as e:
print(f"✗ JSON格式地形加载失败: {e}")
return None
def _load_pickle(self, filename: str) -> Optional[Dict[str, Any]]:
"""
从Pickle格式加载地形数据
Args:
filename: 文件名
Returns:
地形数据字典或None
"""
try:
# 读取文件
with open(filename, 'rb') as f:
file_data = f.read()
# 检查是否压缩
try:
# 尝试解压
decompressed_data = zlib.decompress(file_data)
pickle_data = decompressed_data
self.stats['data_compressed'] += len(file_data)
self.stats['data_uncompressed'] += len(decompressed_data)
except zlib.error:
# 未压缩的Pickle
pickle_data = file_data
self.stats['data_uncompressed'] += len(file_data)
# 解析Pickle
data = pickle.loads(pickle_data)
return data
except Exception as e:
print(f"✗ Pickle格式地形加载失败: {e}")
return None
def _load_binary(self, filename: str) -> Optional[Dict[str, Any]]:
"""
从二进制格式加载地形数据
Args:
filename: 文件名
Returns:
地形数据字典或None
"""
try:
# 读取文件
with open(filename, 'rb') as f:
file_data = f.read()
# 检查是否压缩
try:
# 尝试解压
decompressed_data = zlib.decompress(file_data)
binary_data = decompressed_data
self.stats['data_compressed'] += len(file_data)
self.stats['data_uncompressed'] += len(decompressed_data)
except zlib.error:
# 未压缩的二进制数据
binary_data = file_data
self.stats['data_uncompressed'] += len(file_data)
# 解析二进制数据
data = self._convert_from_binary(binary_data)
return data
except Exception as e:
print(f"✗ 二进制格式地形加载失败: {e}")
return None
def _load_custom(self, filename: str) -> Optional[Dict[str, Any]]:
"""
从自定义格式加载地形数据
Args:
filename: 文件名
Returns:
地形数据字典或None
"""
try:
# 自定义格式加载这里简化为JSON格式
return self._load_json(filename)
except Exception as e:
print(f"✗ 自定义格式地形加载失败: {e}")
return None
def _convert_to_binary(self, data: Dict[str, Any]) -> bytes:
"""
将数据转换为二进制格式
Args:
data: 数据字典
Returns:
二进制数据
"""
try:
# 这里简化为使用Pickle作为二进制格式
return pickle.dumps(data)
except Exception as e:
print(f"✗ 数据转换为二进制失败: {e}")
return b''
def _convert_from_binary(self, binary_data: bytes) -> Dict[str, Any]:
"""
从二进制数据转换为字典
Args:
binary_data: 二进制数据
Returns:
数据字典
"""
try:
# 这里简化为使用Pickle作为二进制格式
return pickle.loads(binary_data)
except Exception as e:
print(f"✗ 二进制数据转换失败: {e}")
return {}
def _detect_format(self, filename: str) -> str:
"""
检测文件格式
Args:
filename: 文件名
Returns:
检测到的格式
"""
try:
# 根据文件扩展名检测格式
if filename.endswith('.json') or filename.endswith('.json.gz'):
return 'json'
elif filename.endswith('.pkl') or filename.endswith('.pkl.gz'):
return 'pickle'
elif filename.endswith('.bin') or filename.endswith('.bin.gz'):
return 'binary'
else:
# 默认使用JSON格式
return self.serialization_params['default_format']
except Exception as e:
print(f"✗ 格式检测失败: {e}")
return self.serialization_params['default_format']
def get_stats(self) -> Dict[str, Any]:
"""
获取统计信息
Returns:
统计信息字典
"""
# 更新平均时间
if self.stats['terrains_saved'] > 0:
self.stats['average_save_time'] = self.stats['total_save_time'] / self.stats['terrains_saved']
if self.stats['terrains_loaded'] > 0:
self.stats['average_load_time'] = self.stats['total_load_time'] / self.stats['terrains_loaded']
# 计算压缩率
total_data = self.stats['data_compressed'] + self.stats['data_uncompressed']
compression_ratio = 0.0
if total_data > 0:
compression_ratio = self.stats['data_compressed'] / total_data if self.stats['data_compressed'] > 0 else 1.0
stats = self.stats.copy()
stats['compression_ratio'] = compression_ratio
return stats
def reset_stats(self):
"""重置统计信息"""
self.stats = {
'terrains_saved': 0,
'terrains_loaded': 0,
'total_save_time': 0.0,
'total_load_time': 0.0,
'average_save_time': 0.0,
'average_load_time': 0.0,
'data_compressed': 0,
'data_uncompressed': 0
}
print("✓ 地形序列化器统计信息已重置")
def set_serialization_parameters(self, params: Dict[str, Any]):
"""
设置序列化参数
Args:
params: 参数字典
"""
self.serialization_params.update(params)
print(f"✓ 序列化参数已更新: {self.serialization_params}")
def export_terrain_to_image(self, heightmap: np.ndarray, filename: str) -> bool:
"""
导出地形为图像文件
Args:
heightmap: 高度图数据
filename: 文件名
Returns:
是否导出成功
"""
try:
print(f"✓ 开始导出地形到图像 {filename}...")
# 确保高度图在0-1范围内
normalized_heightmap = (heightmap - np.min(heightmap)) / (np.max(heightmap) - np.min(heightmap))
# 转换为0-255范围的整数
image_data = (normalized_heightmap * 255).astype(np.uint8)
# 保存为PNG文件
try:
from PIL import Image
image = Image.fromarray(image_data, mode='L')
image.save(filename)
print("✓ 地形图像导出完成")
return True
except ImportError:
# 如果没有PIL保存为原始二进制数据
with open(filename + '.raw', 'wb') as f:
f.write(image_data.tobytes())
print("✓ 地形数据导出完成无PIL保存为.raw文件")
return True
except Exception as e:
print(f"✗ 地形图像导出失败: {e}")
import traceback
traceback.print_exc()
return False
def import_terrain_from_image(self, filename: str) -> Optional[np.ndarray]:
"""
从图像文件导入地形
Args:
filename: 文件名
Returns:
高度图数据或None
"""
try:
print(f"✓ 开始从图像 {filename} 导入地形...")
# 读取图像文件
try:
from PIL import Image
image = Image.open(filename)
# 转换为灰度图
image = image.convert('L')
# 转换为numpy数组
image_data = np.array(image, dtype=np.float32)
# 归一化到0-1范围
heightmap = image_data / 255.0
print("✓ 地形图像导入完成")
return heightmap
except ImportError:
# 如果没有PIL尝试读取原始二进制数据
if filename.endswith('.raw'):
with open(filename, 'rb') as f:
raw_data = f.read()
# 假设是512x512的图像
image_data = np.frombuffer(raw_data, dtype=np.uint8)
heightmap = image_data.astype(np.float32) / 255.0
# 重塑为方形
size = int(np.sqrt(len(heightmap)))
heightmap = heightmap.reshape((size, size))
print("✓ 地形数据导入完成(从.raw文件")
return heightmap
else:
print("✗ 无法读取图像文件需要PIL库")
return None
except Exception as e:
print(f"✗ 地形图像导入失败: {e}")
import traceback
traceback.print_exc()
return None
def backup_terrain(self, filename: str) -> bool:
"""
备份当前地形
Args:
filename: 备份文件名
Returns:
是否备份成功
"""
try:
if not self.enabled:
print("✗ 地形序列化器未启用")
return False
# 获取当前地形数据
if self.plugin.terrain_manager:
terrain_data = self.plugin.terrain_manager.get_terrain_data()
if terrain_data:
# 保存备份
backup_filename = filename + '.backup'
success = self.save_terrain(backup_filename, terrain_data)
if success:
print(f"✓ 地形已备份到 {backup_filename}")
return True
else:
print("✗ 地形备份失败")
return False
else:
print("✗ 无法获取当前地形数据")
return False
else:
print("✗ 地形管理器不可用")
return False
except Exception as e:
print(f"✗ 地形备份失败: {e}")
import traceback
traceback.print_exc()
return False
def restore_terrain(self, filename: str) -> bool:
"""
恢复地形备份
Args:
filename: 备份文件名
Returns:
是否恢复成功
"""
try:
if not self.enabled:
print("✗ 地形序列化器未启用")
return False
# 加载备份数据
backup_filename = filename + '.backup'
terrain_data = self.load_terrain(backup_filename)
if terrain_data:
# 应用到地形管理器
if self.plugin.terrain_manager:
# 这里应该调用地形管理器的方法来恢复数据
# 由于地形管理器可能没有相应的方法,我们只打印信息
print(f"✓ 地形已从 {backup_filename} 恢复")
print("注意:需要在地形管理器中实现实际的恢复逻辑")
return True
else:
print("✗ 地形管理器不可用")
return False
else:
print("✗ 无法加载备份数据")
return False
except Exception as e:
print(f"✗ 地形恢复失败: {e}")
import traceback
traceback.print_exc()
return False
def get_file_info(self, filename: str) -> Dict[str, Any]:
"""
获取文件信息
Args:
filename: 文件名
Returns:
文件信息字典
"""
try:
if not os.path.exists(filename):
return {'error': '文件不存在'}
# 获取文件统计信息
stat = os.stat(filename)
file_info = {
'filename': filename,
'size': stat.st_size,
'created': time.ctime(stat.st_ctime),
'modified': time.ctime(stat.st_mtime),
'accessed': time.ctime(stat.st_atime)
}
# 尝试加载文件以获取地形信息
try:
terrain_data = self.load_terrain(filename)
if terrain_data:
file_info['terrain_info'] = {
'has_heightmap': 'heightmap' in terrain_data and terrain_data['heightmap'] is not None,
'has_biome_map': 'biome_map' in terrain_data and terrain_data['biome_map'] is not None,
'has_vegetation_map': 'vegetation_map' in terrain_data and terrain_data['vegetation_map'] is not None,
'has_water_map': 'water_map' in terrain_data and terrain_data['water_map'] is not None
}
if 'stats' in terrain_data:
file_info['terrain_stats'] = terrain_data['stats']
except Exception as e:
file_info['load_error'] = str(e)
return file_info
except Exception as e:
print(f"✗ 文件信息获取失败: {e}")
return {'error': str(e)}
def list_terrain_files(self, directory: str = '.') -> List[str]:
"""
列出目录中的地形文件
Args:
directory: 目录路径
Returns:
地形文件列表
"""
try:
terrain_files = []
# 支持的文件扩展名
supported_extensions = ['.json', '.pkl', '.bin', '.raw']
# 遍历目录
for filename in os.listdir(directory):
file_path = os.path.join(directory, filename)
if os.path.isfile(file_path):
# 检查文件扩展名
_, ext = os.path.splitext(filename)
if ext in supported_extensions:
terrain_files.append(filename)
return terrain_files
except Exception as e:
print(f"✗ 地形文件列表获取失败: {e}")
return []
def delete_terrain_file(self, filename: str) -> bool:
"""
删除地形文件
Args:
filename: 文件名
Returns:
是否删除成功
"""
try:
if os.path.exists(filename):
os.remove(filename)
print(f"✓ 地形文件 {filename} 已删除")
return True
else:
print(f"✗ 地形文件不存在: {filename}")
return False
except Exception as e:
print(f"✗ 地形文件删除失败: {e}")
return False
def compress_terrain_file(self, filename: str, output_filename: str = None) -> bool:
"""
压缩地形文件
Args:
filename: 输入文件名
output_filename: 输出文件名如果为None则在原文件名后添加.gz
Returns:
是否压缩成功
"""
try:
if not os.path.exists(filename):
print(f"✗ 文件不存在: {filename}")
return False
if output_filename is None:
output_filename = filename + '.gz'
# 读取原始文件
with open(filename, 'rb') as f:
data = f.read()
# 压缩数据
compressed_data = zlib.compress(data, self.serialization_params['compression_level'])
# 写入压缩文件
with open(output_filename, 'wb') as f:
f.write(compressed_data)
print(f"✓ 地形文件已压缩: {filename} -> {output_filename}")
return True
except Exception as e:
print(f"✗ 地形文件压缩失败: {e}")
return False
def decompress_terrain_file(self, filename: str, output_filename: str = None) -> bool:
"""
解压地形文件
Args:
filename: 输入文件名(应该以.gz结尾
output_filename: 输出文件名如果为None则移除.gz后缀
Returns:
是否解压成功
"""
try:
if not os.path.exists(filename):
print(f"✗ 文件不存在: {filename}")
return False
if output_filename is None:
if filename.endswith('.gz'):
output_filename = filename[:-3]
else:
output_filename = filename + '.decompressed'
# 读取压缩文件
with open(filename, 'rb') as f:
compressed_data = f.read()
# 解压数据
try:
data = zlib.decompress(compressed_data)
except zlib.error as e:
print(f"✗ 文件解压失败: {e}")
return False
# 写入解压文件
with open(output_filename, 'wb') as f:
f.write(data)
print(f"✓ 地形文件已解压: {filename} -> {output_filename}")
return True
except Exception as e:
print(f"✗ 地形文件解压失败: {e}")
return False
def validate_terrain_file(self, filename: str) -> bool:
"""
验证地形文件完整性
Args:
filename: 文件名
Returns:
文件是否完整
"""
try:
if not os.path.exists(filename):
print(f"✗ 文件不存在: {filename}")
return False
# 尝试加载文件
terrain_data = self.load_terrain(filename)
if terrain_data is None:
print(f"✗ 地形文件无效: {filename}")
return False
# 检查必要数据是否存在
required_keys = ['heightmap']
missing_keys = [key for key in required_keys if key not in terrain_data or terrain_data[key] is None]
if missing_keys:
print(f"✗ 地形文件缺少必要数据: {missing_keys}")
return False
print(f"✓ 地形文件验证通过: {filename}")
return True
except Exception as e:
print(f"✗ 地形文件验证失败: {e}")
return False
def compare_terrain_files(self, filename1: str, filename2: str) -> Dict[str, Any]:
"""
比较两个地形文件
Args:
filename1: 第一个文件名
filename2: 第二个文件名
Returns:
比较结果字典
"""
try:
# 加载两个文件
terrain1 = self.load_terrain(filename1)
terrain2 = self.load_terrain(filename2)
if terrain1 is None or terrain2 is None:
return {'error': '无法加载一个或两个地形文件'}
comparison = {
'file1': filename1,
'file2': filename2,
'identical': False,
'differences': []
}
# 比较高度图
if ('heightmap' in terrain1 and terrain1['heightmap'] is not None and
'heightmap' in terrain2 and terrain2['heightmap'] is not None):
heightmap1 = np.array(terrain1['heightmap'])
heightmap2 = np.array(terrain2['heightmap'])
if heightmap1.shape == heightmap2.shape:
# 计算差异
diff = np.abs(heightmap1 - heightmap2)
max_diff = np.max(diff)
mean_diff = np.mean(diff)
comparison['heightmap_comparison'] = {
'shapes_match': True,
'max_difference': float(max_diff),
'mean_difference': float(mean_diff),
'identical': max_diff < 1e-6
}
if max_diff >= 1e-6:
comparison['differences'].append('heightmap')
else:
comparison['heightmap_comparison'] = {
'shapes_match': False
}
comparison['differences'].append('heightmap_shape')
else:
comparison['heightmap_comparison'] = {
'missing_data': True
}
# 比较其他数据
maps_to_compare = ['biome_map', 'vegetation_map', 'water_map']
for map_name in maps_to_compare:
if (map_name in terrain1 and terrain1[map_name] is not None and
map_name in terrain2 and terrain2[map_name] is not None):
map1 = np.array(terrain1[map_name])
map2 = np.array(terrain2[map_name])
if map1.shape == map2.shape:
if not np.array_equal(map1, map2):
comparison['differences'].append(map_name)
else:
comparison['differences'].append(f'{map_name}_shape')
# 检查是否完全相同
comparison['identical'] = len(comparison['differences']) == 0
return comparison
except Exception as e:
print(f"✗ 地形文件比较失败: {e}")
return {'error': str(e)}
def merge_terrain_files(self, filenames: List[str], output_filename: str) -> bool:
"""
合并多个地形文件
Args:
filenames: 文件名列表
output_filename: 输出文件名
Returns:
是否合并成功
"""
try:
if len(filenames) < 2:
print("✗ 至少需要两个文件才能合并")
return False
print(f"✓ 开始合并 {len(filenames)} 个地形文件...")
# 加载所有文件
terrains = []
for filename in filenames:
terrain = self.load_terrain(filename)
if terrain is None:
print(f"✗ 无法加载文件: {filename}")
return False
terrains.append(terrain)
# 创建合并后的地形数据
merged_terrain = {}
# 合并高度图(取平均值)
heightmaps = [t.get('heightmap') for t in terrains if t.get('heightmap') is not None]
if heightmaps:
# 转换为numpy数组
np_heightmaps = [np.array(h) for h in heightmaps]
# 确保所有高度图尺寸相同
if all(h.shape == np_heightmaps[0].shape for h in np_heightmaps):
# 计算平均值
merged_heightmap = np.mean(np_heightmaps, axis=0)
merged_terrain['heightmap'] = merged_heightmap
else:
print("✗ 高度图尺寸不匹配,无法合并")
return False
# 合并其他数据(这里简化为取第一个文件的数据)
if terrains:
first_terrain = terrains[0]
for key in ['biome_map', 'vegetation_map', 'water_map', 'stats']:
if key in first_terrain and first_terrain[key] is not None:
merged_terrain[key] = first_terrain[key]
# 保存合并后的地形
success = self.save_terrain(output_filename, merged_terrain)
if success:
print(f"✓ 地形文件合并完成: {output_filename}")
return True
else:
print("✗ 地形文件合并失败")
return False
except Exception as e:
print(f"✗ 地形文件合并失败: {e}")
import traceback
traceback.print_exc()
return False
def convert_terrain_format(self, input_filename: str, output_filename: str,
output_format: str) -> bool:
"""
转换地形文件格式
Args:
input_filename: 输入文件名
output_filename: 输出文件名
output_format: 输出格式
Returns:
是否转换成功
"""
try:
# 验证输出格式
if output_format not in self.supported_formats:
print(f"✗ 不支持的输出格式: {output_format}")
return False
print(f"✓ 开始转换地形格式: {input_filename} -> {output_filename} ({output_format})")
# 加载输入文件
terrain_data = self.load_terrain(input_filename)
if terrain_data is None:
print("✗ 无法加载输入文件")
return False
# 保存为新格式
original_format = self.serialization_params['default_format']
self.serialization_params['default_format'] = output_format
success = self.save_terrain(output_filename, terrain_data, format=output_format)
# 恢复原始格式设置
self.serialization_params['default_format'] = original_format
if success:
print("✓ 地形格式转换完成")
return True
else:
print("✗ 地形格式转换失败")
return False
except Exception as e:
print(f"✗ 地形格式转换失败: {e}")
import traceback
traceback.print_exc()
return False