""" 地形序列化器 负责地形数据的序列化和反序列化 """ 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