""" 侵蚀处理器 负责模拟地形的侵蚀过程 """ import numpy as np import math from typing import Dict, Any import time class ErosionProcessor: """ 侵蚀处理器 负责模拟地形的侵蚀过程，包括水力侵蚀、热力侵蚀、风力侵蚀等 """ def __init__(self, plugin): """ 初始化侵蚀处理器 Args: plugin: 程序化地形生成插件实例 """ self.plugin = plugin self.enabled = False self.initialized = False # 侵蚀配置 self.seed = plugin.config.get('seed', 12345) # 侵蚀参数 self.erosion_params = { 'water_erosion': { 'enabled': True, 'rate': 0.1, 'solubility': 0.8, 'evaporation': 0.01, 'iterations': 50, 'inertia': 0.05, 'sediment_capacity': 0.5, 'deposition_rate': 0.1 }, 'thermal_erosion': { 'enabled': True, 'rate': 0.05, 'angle_of_repose': 30.0, # 度 'iterations': 30 }, 'wind_erosion': { 'enabled': True, 'rate': 0.02, 'direction': 0.0, # 弧度 'iterations': 20 }, 'chemical_erosion': { 'enabled': False, 'rate': 0.01, 'iterations': 10 }, 'glacial_erosion': { 'enabled': False, 'rate': 0.03, 'iterations': 15 } } # 侵蚀图层 self.erosion_map = None self.sediment_map = None self.water_map = None self.flow_map = None # 统计信息 self.stats = { 'erosion_processed': 0, 'total_processing_time': 0.0, 'average_processing_time': 0.0, 'total_erosion_amount': 0.0, 'total_sediment_amount': 0.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 apply_erosion(self, heightmap: np.ndarray, seed: int = None) -> Dict[str, np.ndarray]: """ 应用侵蚀效果到高度图 Args: heightmap: 高度图数据 seed: 随机种子 Returns: 包含处理后高度图和侵蚀相关数据的字典 """ try: processing_start_time = time.time() if not self.enabled: print("✗ 侵蚀处理器未启用") return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } # 更新种子 if seed is not None: self.seed = seed print("✓ 开始应用侵蚀效果...") # 初始化侵蚀图层 self._initialize_erosion_layers(heightmap.shape) # 创建高度图副本进行处理 processed_heightmap = heightmap.copy() # 应用各种侵蚀类型 if self.erosion_params['water_erosion']['enabled']: print(" → 应用水力侵蚀...") processed_heightmap = self._apply_water_erosion(processed_heightmap) if self.erosion_params['thermal_erosion']['enabled']: print(" → 应用热力侵蚀...") processed_heightmap = self._apply_thermal_erosion(processed_heightmap) if self.erosion_params['wind_erosion']['enabled']: print(" → 应用风力侵蚀...") processed_heightmap = self._apply_wind_erosion(processed_heightmap) if self.erosion_params['chemical_erosion']['enabled']: print(" → 应用化学侵蚀...") processed_heightmap = self._apply_chemical_erosion(processed_heightmap) if self.erosion_params['glacial_erosion']['enabled']: print(" → 应用冰川侵蚀...") processed_heightmap = self._apply_glacial_erosion(processed_heightmap) # 更新统计信息 processing_time = time.time() - processing_start_time self.stats['erosion_processed'] += 1 self.stats['total_processing_time'] += processing_time self.stats['average_processing_time'] = self.stats['total_processing_time'] / self.stats['erosion_processed'] print(f"✓ 侵蚀处理完成，耗时: {processing_time:.2f}秒") return { 'heightmap': processed_heightmap, 'erosion_map': self.erosion_map, 'sediment_map': self.sediment_map, 'water_map': self.water_map } except Exception as e: print(f"✗ 侵蚀处理失败: {e}") import traceback traceback.print_exc() return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } def _initialize_erosion_layers(self, shape: tuple): """初始化侵蚀相关图层""" try: height, width = shape self.erosion_map = np.zeros((height, width), dtype=np.float32) self.sediment_map = np.zeros((height, width), dtype=np.float32) self.water_map = np.zeros((height, width), dtype=np.float32) self.flow_map = np.zeros((height, width, 2), dtype=np.float32) # x, y 流向 except Exception as e: print(f"✗ 侵蚀图层初始化失败: {e}") def _apply_water_erosion(self, heightmap: np.ndarray) -> np.ndarray: """ 应用水力侵蚀 Args: heightmap: 高度图数据 Returns: 处理后的高度图 """ try: processed = heightmap.copy() height, width = processed.shape # 获取水力侵蚀参数 params = self.erosion_params['water_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] solubility = params['solubility'] evaporation = params['evaporation'] inertia = params['inertia'] sediment_capacity = params['sediment_capacity'] deposition_rate = params['deposition_rate'] # 模拟多个迭代步骤 for iteration in range(iterations): # 计算水流方向 self._calculate_water_flow(processed) # 模拟侵蚀和沉积 for y in range(height): for x in range(width): # 计算当前点的侵蚀量 erosion_amount = erosion_rate * solubility * self.water_map[y, x] # 应用侵蚀 if erosion_amount > 0: processed[y, x] -= erosion_amount self.erosion_map[y, x] += erosion_amount self.sediment_map[y, x] += erosion_amount # 检查沉积（如果水流速度慢且携带沉积物过多） sediment_ratio = self.sediment_map[y, x] / max(0.001, self.water_map[y, x]) if sediment_ratio > sediment_capacity: deposition_amount = (sediment_ratio - sediment_capacity) * deposition_rate processed[y, x] += deposition_amount self.sediment_map[y, x] -= deposition_amount # 应用蒸发 self.water_map *= (1.0 - evaporation) # 确保高度值合理 processed = np.clip(processed, 0.0, 1.0) # 更新统计信息 self.stats['total_erosion_amount'] += np.sum(self.erosion_map) self.stats['total_sediment_amount'] += np.sum(self.sediment_map) return processed except Exception as e: print(f"✗ 水力侵蚀应用失败: {e}") return heightmap def _calculate_water_flow(self, heightmap: np.ndarray): """ 计算水流方向和累积 Args: heightmap: 高度图数据 """ try: height, width = heightmap.shape # 简化的水流计算 for y in range(height): for x in range(width): # 增加降雨量 self.water_map[y, x] += 0.01 # 计算流向（流向最低的邻居） lowest_height = heightmap[y, x] flow_x, flow_y = 0, 0 # 检查8个邻居 for dy in [-1, 0, 1]: for dx in [-1, 0, 1]: if dy == 0 and dx == 0: continue ny, nx = y + dy, x + dx if 0 <= ny < height and 0 <= nx < width: neighbor_height = heightmap[ny, nx] + self.water_map[ny, nx] if neighbor_height < lowest_height: lowest_height = neighbor_height flow_x, flow_y = dx, dy # 设置流向 self.flow_map[y, x, 0] = flow_x self.flow_map[y, x, 1] = flow_y # 将水流移动到下一个位置 if flow_x != 0 or flow_y != 0: ny, nx = y + int(flow_y), x + int(flow_x) if 0 <= ny < height and 0 <= nx < width: self.water_map[ny, nx] += self.water_map[y, x] * 0.8 self.water_map[y, x] *= 0.2 except Exception as e: print(f"✗ 水流计算失败: {e}") def _apply_thermal_erosion(self, heightmap: np.ndarray) -> np.ndarray: """ 应用热力侵蚀（重力侵蚀） Args: heightmap: 高度图数据 Returns: 处理后的高度图 """ try: processed = heightmap.copy() height, width = processed.shape # 获取热力侵蚀参数 params = self.erosion_params['thermal_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] angle_of_repose = math.radians(params['angle_of_repose']) tan_repose = math.tan(angle_of_repose) # 模拟多个迭代步骤 for iteration in range(iterations): new_heightmap = processed.copy() # 对每个点应用热力侵蚀 for y in range(height): for x in range(width): current_height = processed[y, x] # 检查4个直接邻居 for dy, dx in [(-1, 0), (1, 0), (0, -1), (0, 1)]: ny, nx = y + dy, x + dx if 0 <= ny < height and 0 <= nx < width: neighbor_height = processed[ny, nx] # 计算高度差 height_diff = current_height - neighbor_height # 如果高度差超过角度限制 if height_diff > 0: # 计算最大稳定高度差 max_diff = tan_repose * 1.0 # 假设网格间距为1 # 如果超过最大稳定差值，则发生侵蚀 if height_diff > max_diff: # 移动材料到较低位置 move_amount = (height_diff - max_diff) * erosion_rate new_heightmap[y, x] -= move_amount new_heightmap[ny, nx] += move_amount # 更新侵蚀图 self.erosion_map[y, x] += move_amount processed = new_heightmap return processed except Exception as e: print(f"✗ 热力侵蚀应用失败: {e}") return heightmap def _apply_wind_erosion(self, heightmap: np.ndarray) -> np.ndarray: """ 应用风力侵蚀 Args: heightmap: 高度图数据 Returns: 处理后的高度图 """ try: processed = heightmap.copy() height, width = processed.shape # 获取风力侵蚀参数 params = self.erosion_params['wind_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] wind_direction = params['direction'] # 计算风向向量 wind_dx = math.cos(wind_direction) wind_dy = math.sin(wind_direction) # 模拟多个迭代步骤 for iteration in range(iterations): new_heightmap = processed.copy() # 对每个点应用风力侵蚀 for y in range(height): for x in range(width): # 风从特定方向吹来 # 计算风的来源位置 source_x = int(x - wind_dx * 2) # 2个单位距离 source_y = int(y - wind_dy * 2) # 如果来源位置有效 if 0 <= source_y < height and 0 <= source_x < width: # 如果当前点比来源点高，则可能发生风蚀 height_diff = processed[y, x] - processed[source_y, source_x] if height_diff > 0.01: # 阈值 # 应用风力侵蚀 erosion_amount = erosion_rate * height_diff new_heightmap[y, x] -= erosion_amount self.erosion_map[y, x] += erosion_amount processed = new_heightmap return processed except Exception as e: print(f"✗ 风力侵蚀应用失败: {e}") return heightmap def _apply_chemical_erosion(self, heightmap: np.ndarray) -> np.ndarray: """ 应用化学侵蚀 Args: heightmap: 高度图数据 Returns: 处理后的高度图 """ try: processed = heightmap.copy() height, width = processed.shape # 获取化学侵蚀参数 params = self.erosion_params['chemical_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] # 模拟多个迭代步骤 for iteration in range(iterations): # 简化的化学侵蚀：基于高度和随机因素 for y in range(height): for x in range(width): # 化学侵蚀通常影响较低的区域（如河流、湖泊） if processed[y, x] < 0.3: # 假设低地更容易发生化学侵蚀 # 添加随机因素 chemical_factor = np.random.random() * 0.5 + 0.5 erosion_amount = erosion_rate * chemical_factor * (0.3 - processed[y, x]) processed[y, x] -= erosion_amount self.erosion_map[y, x] += erosion_amount return processed except Exception as e: print(f"✗ 化学侵蚀应用失败: {e}") return heightmap def _apply_glacial_erosion(self, heightmap: np.ndarray) -> np.ndarray: """ 应用冰川侵蚀 Args: heightmap: 高度图数据 Returns: 处理后的高度图 """ try: processed = heightmap.copy() height, width = processed.shape # 获取冰川侵蚀参数 params = self.erosion_params['glacial_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] # 模拟多个迭代步骤 for iteration in range(iterations): # 冰川侵蚀：平滑地形并创建U型谷 new_heightmap = processed.copy() for y in range(height): for x in range(width): # 冰川通常在高海拔地区形成 if processed[y, x] > 0.7: # 高海拔 # 平滑周围区域 total_height = 0.0 count = 0 # 检查3x3邻域 for dy in [-1, 0, 1]: for dx in [-1, 0, 1]: ny, nx = y + dy, x + dx if 0 <= ny < height and 0 <= nx < width: total_height += processed[ny, nx] count += 1 if count > 0: average_height = total_height / count # 应用冰川侵蚀（平滑化） glacial_effect = erosion_rate * (processed[y, x] - average_height) new_heightmap[y, x] -= glacial_effect self.erosion_map[y, x] += abs(glacial_effect) processed = new_heightmap return processed except Exception as e: print(f"✗ 冰川侵蚀应用失败: {e}") return heightmap def set_seed(self, seed: int): """ 设置随机种子 Args: seed: 随机种子 """ self.seed = seed print(f"✓ 侵蚀处理器随机种子设置为: {seed}") def set_erosion_parameters(self, erosion_type: str, params: Dict[str, Any]): """ 设置侵蚀参数 Args: erosion_type: 侵蚀类型 params: 参数字典 """ if erosion_type in self.erosion_params: self.erosion_params[erosion_type].update(params) print(f"✓ {erosion_type} 侵蚀参数已更新: {self.erosion_params[erosion_type]}") else: print(f"✗ 无效的侵蚀类型: {erosion_type}") def enable_erosion_type(self, erosion_type: str, enabled: bool): """ 启用/禁用侵蚀类型 Args: erosion_type: 侵蚀类型 enabled: 是否启用 """ if erosion_type in self.erosion_params: self.erosion_params[erosion_type]['enabled'] = enabled state = "启用" if enabled else "禁用" print(f"✓ {erosion_type} 侵蚀已{state}") else: print(f"✗ 无效的侵蚀类型: {erosion_type}") def get_stats(self) -> Dict[str, Any]: """ 获取统计信息 Returns: 统计信息字典 """ # 更新平均处理时间 if self.stats['erosion_processed'] > 0: self.stats['average_processing_time'] = self.stats['total_processing_time'] / self.stats['erosion_processed'] return self.stats.copy() def reset_stats(self): """重置统计信息""" self.stats = { 'erosion_processed': 0, 'total_processing_time': 0.0, 'average_processing_time': 0.0, 'total_erosion_amount': 0.0, 'total_sediment_amount': 0.0 } print("✓ 侵蚀处理器统计信息已重置") def get_erosion_map(self) -> np.ndarray: """ 获取侵蚀图 Returns: 侵蚀图数据 """ return self.erosion_map if self.erosion_map is not None else np.array([]) def get_sediment_map(self) -> np.ndarray: """ 获取沉积物图 Returns: 沉积物图数据 """ return self.sediment_map if self.sediment_map is not None else np.array([]) def get_water_map(self) -> np.ndarray: """ 获取水图 Returns: 水图数据 """ return self.water_map if self.water_map is not None else np.array([]) def set_water_erosion_parameters(self, rate: float = None, solubility: float = None, evaporation: float = None, inertia: float = None, sediment_capacity: float = None, deposition_rate: float = None): """ 设置水力侵蚀参数 Args: rate: 侵蚀速率 solubility: 溶解度 evaporation: 蒸发率 inertia: 惯性 sediment_capacity: 沉积物容量 deposition_rate: 沉积率 """ params = self.erosion_params['water_erosion'] if rate is not None: params['rate'] = max(0.0, rate) if solubility is not None: params['solubility'] = max(0.0, min(1.0, solubility)) if evaporation is not None: params['evaporation'] = max(0.0, min(1.0, evaporation)) if inertia is not None: params['inertia'] = max(0.0, min(1.0, inertia)) if sediment_capacity is not None: params['sediment_capacity'] = max(0.0, sediment_capacity) if deposition_rate is not None: params['deposition_rate'] = max(0.0, deposition_rate) print(f"✓ 水力侵蚀参数已更新: {params}") def set_thermal_erosion_parameters(self, rate: float = None, angle_of_repose: float = None): """ 设置热力侵蚀参数 Args: rate: 侵蚀速率 angle_of_repose: 休止角（度） """ params = self.erosion_params['thermal_erosion'] if rate is not None: params['rate'] = max(0.0, rate) if angle_of_repose is not None: params['angle_of_repose'] = max(0.0, angle_of_repose) print(f"✓ 热力侵蚀参数已更新: {params}") def set_wind_erosion_parameters(self, rate: float = None, direction: float = None): """ 设置风力侵蚀参数 Args: rate: 侵蚀速率 direction: 风向（弧度） """ params = self.erosion_params['wind_erosion'] if rate is not None: params['rate'] = max(0.0, rate) if direction is not None: params['direction'] = direction print(f"✓ 风力侵蚀参数已更新: {params}") def apply_selective_erosion(self, heightmap: np.ndarray, mask: np.ndarray, seed: int = None) -> Dict[str, np.ndarray]: """ 在指定区域应用侵蚀 Args: heightmap: 高度图数据 mask: 掩码（0-1范围，1表示完全应用侵蚀） seed: 随机种子 Returns: 包含处理后高度图和侵蚀相关数据的字典 """ try: processing_start_time = time.time() if not self.enabled: print("✗ 侵蚀处理器未启用") return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } # 更新种子 if seed is not None: self.seed = seed print("✓ 开始应用选择性侵蚀...") # 确保掩码尺寸匹配 if mask.shape != heightmap.shape: print("✗ 掩码尺寸与高度图不匹配") return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } # 初始化侵蚀图层 self._initialize_erosion_layers(heightmap.shape) # 创建高度图副本进行处理 processed_heightmap = heightmap.copy() # 在掩码区域应用侵蚀 mask_indices = np.where(mask > 0.5) # 创建临时高度图只包含掩码区域 temp_heightmap = heightmap.copy() # 应用各种侵蚀类型（只在掩码区域） if self.erosion_params['water_erosion']['enabled']: print(" → 应用水力侵蚀（选择性）...") temp_heightmap = self._apply_water_erosion_selective(temp_heightmap, mask) if self.erosion_params['thermal_erosion']['enabled']: print(" → 应用热力侵蚀（选择性）...") temp_heightmap = self._apply_thermal_erosion_selective(temp_heightmap, mask) if self.erosion_params['wind_erosion']['enabled']: print(" → 应用风力侵蚀（选择性）...") temp_heightmap = self._apply_wind_erosion_selective(temp_heightmap, mask) # 只更新掩码区域的值 processed_heightmap[mask_indices] = temp_heightmap[mask_indices] # 更新统计信息 processing_time = time.time() - processing_start_time self.stats['erosion_processed'] += 1 self.stats['total_processing_time'] += processing_time self.stats['average_processing_time'] = self.stats['total_processing_time'] / self.stats['erosion_processed'] print(f"✓ 选择性侵蚀处理完成，耗时: {processing_time:.2f}秒") return { 'heightmap': processed_heightmap, 'erosion_map': self.erosion_map, 'sediment_map': self.sediment_map, 'water_map': self.water_map } except Exception as e: print(f"✗ 选择性侵蚀处理失败: {e}") import traceback traceback.print_exc() return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } def _apply_water_erosion_selective(self, heightmap: np.ndarray, mask: np.ndarray) -> np.ndarray: """在选择区域应用水力侵蚀""" try: processed = heightmap.copy() height, width = processed.shape # 获取水力侵蚀参数 params = self.erosion_params['water_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] * 0.5 # 降低速率以适应选择性应用 # 只在掩码区域应用侵蚀 mask_indices = np.where(mask > 0.5) # 简化的选择性水力侵蚀 for iteration in range(iterations): for y, x in zip(mask_indices[0], mask_indices[1]): # 简单的侵蚀：降低高点，填充低点 if processed[y, x] > 0.8: # 高点 erosion_amount = erosion_rate * mask[y, x] processed[y, x] -= erosion_amount self.erosion_map[y, x] += erosion_amount elif processed[y, x] < 0.2: # 低点 deposition_amount = erosion_rate * 0.5 * mask[y, x] processed[y, x] += deposition_amount self.sediment_map[y, x] += deposition_amount return processed except Exception as e: print(f"✗ 选择性水力侵蚀应用失败: {e}") return heightmap def _apply_thermal_erosion_selective(self, heightmap: np.ndarray, mask: np.ndarray) -> np.ndarray: """在选择区域应用热力侵蚀""" try: processed = heightmap.copy() height, width = processed.shape # 获取热力侵蚀参数 params = self.erosion_params['thermal_erosion'] iterations = params['iterations'] erosion_rate = params['rate'] * 0.5 # 降低速率以适应选择性应用 # 只在掩码区域应用侵蚀 mask_indices = np.where(mask > 0.5) # 简化的选择性热力侵蚀 for iteration in range(iterations): for y, x in zip(mask_indices[0], mask_indices[1]): # 检查直接邻居 for dy, dx in [(-1, 0), (1, 0), (0, -1), (0, 1)]: ny, nx = y + dy, x + dx if 0 <= ny < height and 0 <= nx < width: height_diff = processed[y, x] - processed[ny, nx] if height_diff > 0.1: # 显著高度差 move_amount = erosion_rate * height_diff * mask[y, x] processed[y, x] -= move_amount processed[ny, nx] += move_amount self.erosion_map[y, x] += move_amount return processed except Exception as e: print(f"✗ 选择性热力侵蚀应用失败: {e}") return heightmap def _apply_wind_erosion_selective(self, heightmap: np.ndarray, mask: np.ndarray) -> np.ndarray: """在选择区域应用风力侵蚀""" try: processed = heightmap.copy() height, width = processed.shape # 获取风力侵蚀参数 params = self.erosion_params['wind_erosion'] erosion_rate = params['rate'] * 0.5 # 降低速率以适应选择性应用 wind_direction = params['direction'] # 计算风向向量 wind_dx = math.cos(wind_direction) wind_dy = math.sin(wind_direction) # 只在掩码区域应用侵蚀 mask_indices = np.where(mask > 0.5) # 简化的选择性风力侵蚀 for y, x in zip(mask_indices[0], mask_indices[1]): # 风从特定方向吹来 source_x = int(x - wind_dx * 1) # 1个单位距离 source_y = int(y - wind_dy * 1) # 如果来源位置有效且也在掩码区域内 if (0 <= source_y < height and 0 <= source_x < width and mask[source_y, source_x] > 0.5): # 应用风力侵蚀 height_diff = processed[y, x] - processed[source_y, source_x] if height_diff > 0.01: erosion_amount = erosion_rate * height_diff * mask[y, x] processed[y, x] -= erosion_amount self.erosion_map[y, x] += erosion_amount return processed except Exception as e: print(f"✗ 选择性风力侵蚀应用失败: {e}") return heightmap def blend_erosion_maps(self, erosion_maps: List[np.ndarray], weights: List[float] = None) -> np.ndarray: """ 混合多个侵蚀图 Args: erosion_maps: 侵蚀图列表 weights: 权重列表（如果为None，则平均分配权重） Returns: 混合后的侵蚀图 """ try: if not erosion_maps: return np.array([]) if len(erosion_maps) == 1: return erosion_maps[0].copy() # 确保所有侵蚀图尺寸相同 base_shape = erosion_maps[0].shape for em in erosion_maps: if em.shape != base_shape: raise ValueError("所有侵蚀图必须具有相同的尺寸") # 处理权重 if weights is None: weights = [1.0 / len(erosion_maps)] * len(erosion_maps) elif len(weights) != len(erosion_maps): raise ValueError("权重数量必须与侵蚀图数量相同") # 归一化权重 total_weight = sum(weights) if total_weight > 0: weights = [w / total_weight for w in weights] # 混合侵蚀图 blended = np.zeros(base_shape, dtype=np.float32) for i, (em, weight) in enumerate(zip(erosion_maps, weights)): blended += em * weight return blended except Exception as e: print(f"✗ 侵蚀图混合失败: {e}") import traceback traceback.print_exc() # 返回第一个侵蚀图的副本 return erosion_maps[0].copy() if erosion_maps else np.array([]) def smooth_erosion_map(self, erosion_map: np.ndarray, kernel_size: int = 3) -> np.ndarray: """ 平滑侵蚀图 Args: erosion_map: 原始侵蚀图 kernel_size: 平滑核大小 Returns: 平滑后的侵蚀图 """ try: if kernel_size < 1: return erosion_map smoothed = erosion_map.copy() height, width = smoothed.shape # 简单的均值滤波 half_kernel = kernel_size // 2 for y in range(height): for x in range(width): total = 0.0 count = 0 # 应用平滑核 for dy in range(-half_kernel, half_kernel + 1): for dx in range(-half_kernel, half_kernel + 1): ny, nx = y + dy, x + dx if 0 <= ny < height and 0 <= nx < width: total += erosion_map[ny, nx] count += 1 if count > 0: smoothed[y, x] = total / count return smoothed except Exception as e: print(f"✗ 侵蚀图平滑失败: {e}") return erosion_map def apply_erosion_intensity(self, heightmap: np.ndarray, intensity_map: np.ndarray, seed: int = None) -> Dict[str, np.ndarray]: """ 根据强度图应用侵蚀 Args: heightmap: 高度图数据 intensity_map: 侵蚀强度图（0-1范围） seed: 随机种子 Returns: 包含处理后高度图和侵蚀相关数据的字典 """ try: processing_start_time = time.time() if not self.enabled: print("✗ 侵蚀处理器未启用") return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } # 更新种子 if seed is not None: self.seed = seed # 确保强度图尺寸匹配 if intensity_map.shape != heightmap.shape: print("✗ 强度图尺寸与高度图不匹配") return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } print("✓ 开始应用强度控制的侵蚀...") # 初始化侵蚀图层 self._initialize_erosion_layers(heightmap.shape) # 创建高度图副本进行处理 processed_heightmap = heightmap.copy() # 根据强度图调整侵蚀参数 base_water_params = self.erosion_params['water_erosion'].copy() base_thermal_params = self.erosion_params['thermal_erosion'].copy() base_wind_params = self.erosion_params['wind_erosion'].copy() # 对每个点根据强度应用侵蚀 height, width = processed_heightmap.shape for y in range(height): for x in range(width): intensity = intensity_map[y, x] if intensity > 0.1: # 只在强度大于阈值时应用侵蚀 # 调整侵蚀参数 self.erosion_params['water_erosion']['rate'] = base_water_params['rate'] * intensity self.erosion_params['thermal_erosion']['rate'] = base_thermal_params['rate'] * intensity self.erosion_params['wind_erosion']['rate'] = base_wind_params['rate'] * intensity # 应用局部侵蚀（简化版本） if self.erosion_params['water_erosion']['enabled']: # 简化的水力侵蚀 if processed_heightmap[y, x] > 0.8: erosion_amount = self.erosion_params['water_erosion']['rate'] * 0.1 processed_heightmap[y, x] -= erosion_amount self.erosion_map[y, x] += erosion_amount # 恢复原始侵蚀参数 self.erosion_params['water_erosion'] = base_water_params self.erosion_params['thermal_erosion'] = base_thermal_params self.erosion_params['wind_erosion'] = base_wind_params # 更新统计信息 processing_time = time.time() - processing_start_time self.stats['erosion_processed'] += 1 self.stats['total_processing_time'] += processing_time self.stats['average_processing_time'] = self.stats['total_processing_time'] / self.stats['erosion_processed'] print(f"✓ 强度控制的侵蚀处理完成，耗时: {processing_time:.2f}秒") return { 'heightmap': processed_heightmap, 'erosion_map': self.erosion_map, 'sediment_map': self.sediment_map, 'water_map': self.water_map } except Exception as e: print(f"✗ 强度控制的侵蚀处理失败: {e}") import traceback traceback.print_exc() return { 'heightmap': heightmap, 'erosion_map': np.zeros_like(heightmap, dtype=np.float32), 'sediment_map': np.zeros_like(heightmap, dtype=np.float32), 'water_map': np.zeros_like(heightmap, dtype=np.float32) } def export_erosion_data(self, filename: str) -> bool: """ 导出侵蚀数据 Args: filename: 文件名 Returns: 是否导出成功 """ try: import json erosion_data = { 'parameters': self.erosion_params, 'stats': self.stats, 'timestamp': time.time() } with open(filename, 'w', encoding='utf-8') as f: json.dump(erosion_data, f, ensure_ascii=False, indent=2) print(f"✓ 侵蚀数据已导出到: {filename}") return True except Exception as e: print(f"✗ 侵蚀数据导出失败: {e}") return False def import_erosion_data(self, filename: str) -> bool: """ 导入侵蚀数据 Args: filename: 文件名 Returns: 是否导入成功 """ try: import json with open(filename, 'r', encoding='utf-8') as f: erosion_data = json.load(f) # 更新参数 if 'parameters' in erosion_data: self.erosion_params.update(erosion_data['parameters']) # 更新统计信息 if 'stats' in erosion_data: self.stats.update(erosion_data['stats']) print(f"✓ 侵蚀数据已从 {filename} 导入") return True except Exception as e: print(f"✗ 侵蚀数据导入失败: {e}") return False def reset_erosion_maps(self): """重置侵蚀图层""" self.erosion_map = None self.sediment_map = None self.water_map = None self.flow_map = None print("✓ 侵蚀图层已重置") def get_erosion_types(self) -> List[str]: """ 获取可用的侵蚀类型 Returns: 侵蚀类型列表 """ return list(self.erosion_params.keys()) def get_erosion_parameters(self, erosion_type: str) -> Dict[str, Any]: """ 获取指定侵蚀类型的参数 Args: erosion_type: 侵蚀类型 Returns: 参数字典 """ return self.erosion_params.get(erosion_type, {}).copy()