""" 生态系统交互模块 负责处理植被与动物之间以及生态系统内部的复杂交互 """ import time from typing import Dict, Any, List, Optional import math import random class EcoInteractions: """ 生态系统交互管理器 负责处理植被与动物之间以及生态系统内部的复杂交互 """ def __init__(self, plugin): """ 初始化生态系统交互管理器 Args: plugin: 植被和生态系统插件实例 """ self.plugin = plugin self.enabled = False self.initialized = False # 交互类型定义 self.interaction_types = { 'herbivory': { 'name': '草食行为', 'description': '动物食用植物', 'affects': ['vegetation', 'animal'], 'rate_factor': 1.0 }, 'seed_dispersal': { 'name': '种子传播', 'description': '动物帮助植物传播种子', 'affects': ['vegetation'], 'rate_factor': 0.5 }, 'pollination': { 'name': '授粉', 'description': '动物为植物授粉', 'affects': ['vegetation'], 'rate_factor': 0.8 }, 'habitat_modification': { 'name': '栖息地改造', 'description': '动物活动改变环境', 'affects': ['environment'], 'rate_factor': 0.3 }, 'competition': { 'name': '竞争', 'description': '物种间资源竞争', 'affects': ['vegetation', 'animal'], 'rate_factor': 0.7 }, 'mutualism': { 'name': '互利共生', 'description': '物种间互利关系', 'affects': ['vegetation', 'animal'], 'rate_factor': 0.6 } } # 交互参数 self.interaction_params = { 'herbivory_rate': 0.05, # 草食率 'seed_dispersal_rate': 0.1, # 种子传播率 'pollination_rate': 0.15, # 授粉率 'competition_intensity': 0.2, # 竞争强度 'mutualism_benefit': 0.1 # 互利共生益处 } # 交互效果 self.interaction_effects = { 'vegetation_damage': 0.0, # 植被损害 'vegetation_growth_boost': 0.0, # 植被生长促进 'animal_nutrition': 0.0, # 动物营养 'biodiversity_impact': 0.0 # 生物多样性影响 } # 交互历史 self.interaction_history = [] self.max_history_size = 1000 # 时间配置 self.time_config = { 'update_interval': 60.0, # 更新间隔(秒) 'last_update': 0.0 } # 统计信息 self.interaction_stats = { 'total_interactions': 0, 'herbivory_events': 0, 'seed_dispersal_events': 0, 'pollination_events': 0, 'competition_events': 0, 'mutualism_events': 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.interaction_history.clear() self.initialized = False print("✓ 生态系统交互管理器资源已清理") except Exception as e: print(f"✗ 生态系统交互管理器资源清理失败: {e}") import traceback traceback.print_exc() def update(self, dt: float): """ 更新生态系统交互管理器状态 Args: dt: 时间增量 """ try: if not self.enabled: return # 更新时间 self.time_config['last_update'] += dt # 检查是否需要更新 if self.time_config['last_update'] >= self.time_config['update_interval']: # 处理生态系统交互 self._process_ecosystem_interactions() # 重置更新计时器 self.time_config['last_update'] = 0.0 except Exception as e: print(f"✗ 生态系统交互管理器更新失败: {e}") import traceback traceback.print_exc() def _process_ecosystem_interactions(self): """处理生态系统交互""" try: # 处理草食行为 self._process_herbivory() # 处理种子传播 self._process_seed_dispersal() # 处理授粉 self._process_pollination() # 处理竞争 self._process_competition() # 处理互利共生 self._process_mutualism() # 更新统计信息 self.interaction_stats['total_interactions'] += 1 except Exception as e: print(f"✗ 生态系统交互处理失败: {e}") def _process_herbivory(self): """处理草食行为""" try: if not self.plugin.animal_simulator or not self.plugin.vegetation_manager: return # 获取动物和植被信息 animal_populations = self.plugin.animal_simulator.get_population_data() vegetation_instances = self.plugin.vegetation_manager.get_all_vegetation_instances() # 遍历动物种群 for population_id, population_data in animal_populations.items(): species = population_data['species'] population_size = population_data['size'] # 检查动物是否为草食性 animal_info = self.plugin.animal_simulator.get_animal_info(species) if not animal_info or animal_info['diet'] not in ['herbivore', 'omnivore']: continue # 计算草食事件概率 herbivory_probability = self.interaction_params['herbivory_rate'] * population_size / 100 if random.random() < herbivory_probability: # 选择一个植被实例作为目标 if vegetation_instances: target_instance_id = random.choice(list(vegetation_instances.keys())) target_instance = vegetation_instances[target_instance_id] # 检查动物是否偏好这种植被 preferred_food = animal_info.get('preferred_food', []) if target_instance['type'] in preferred_food: damage_factor = 1.5 # 偏好食物造成更多损害 else: damage_factor = 1.0 # 对植被造成损害 damage = 0.05 * damage_factor self.plugin.vegetation_manager.update_vegetation_health(target_instance_id, -damage) # 动物获得营养 nutrition_gain = 0.02 * damage_factor # 这里可以更新动物状态(如果动物实例系统支持) # 记录交互事件 self._record_interaction_event('herbivory', { 'animal_species': species, 'vegetation_type': target_instance['type'], 'damage': damage, 'nutrition_gain': nutrition_gain }) # 更新统计信息 self.interaction_stats['herbivory_events'] += 1 except Exception as e: print(f"✗ 草食行为处理失败: {e}") def _process_seed_dispersal(self): """处理种子传播""" try: if not self.plugin.animal_simulator or not self.plugin.vegetation_manager: return # 获取动物和植被信息 animal_populations = self.plugin.animal_simulator.get_population_data() vegetation_instances = self.plugin.vegetation_manager.get_all_vegetation_instances() # 遍历动物种群 for population_id, population_data in animal_populations.items(): species = population_data['species'] population_size = population_data['size'] # 检查动物是否适合传播种子(鸟类、哺乳动物等) animal_info = self.plugin.animal_simulator.get_animal_info(species) if not animal_info or animal_info['size'] not in ['small', 'medium', 'large']: continue # 计算种子传播概率 dispersal_probability = self.interaction_params['seed_dispersal_rate'] * population_size / 100 if random.random() < dispersal_probability: # 选择一个成熟的植被实例作为种子来源 mature_vegetation = [ (vid, vdata) for vid, vdata in vegetation_instances.items() if vdata['alive'] and vdata['size'] > 0.7 ] if mature_vegetation: source_instance_id, source_instance = random.choice(mature_vegetation) # 计算新的生成位置(在动物活动范围内) animal_territory = animal_info.get('territory_size', 30.0) source_pos = source_instance['position'] angle = random.uniform(0, 2 * math.pi) distance = random.uniform(0, animal_territory) new_x = source_pos[0] + distance * math.cos(angle) new_z = source_pos[2] + distance * math.sin(angle) # 创建新的植被实例 new_instance_id = self.plugin.vegetation_manager.create_vegetation_instance( source_instance['type'], (new_x, source_pos[1], new_z), initial_age=0.0, initial_health=0.3 ) if new_instance_id >= 0: # 记录交互事件 self._record_interaction_event('seed_dispersal', { 'animal_species': species, 'vegetation_type': source_instance['type'], 'source_position': source_pos, 'new_position': (new_x, source_pos[1], new_z) }) # 更新统计信息 self.interaction_stats['seed_dispersal_events'] += 1 except Exception as e: print(f"✗ 种子传播处理失败: {e}") def _process_pollination(self): """处理授粉""" try: if not self.plugin.animal_simulator or not self.plugin.vegetation_manager: return # 获取动物和植被信息 animal_populations = self.plugin.animal_simulator.get_population_data() vegetation_instances = self.plugin.vegetation_manager.get_all_vegetation_instances() # 遍历动物种群 for population_id, population_data in animal_populations.items(): species = population_data['species'] population_size = population_data['size'] # 检查动物是否适合授粉(昆虫、鸟类等) animal_info = self.plugin.animal_simulator.get_animal_info(species) if not animal_info or 'pollinator' not in animal_info.get('description', ''): # 特殊处理鸟类和昆虫 if species not in ['bird', 'insect']: continue # 计算授粉概率 pollination_probability = self.interaction_params['pollination_rate'] * population_size / 100 if random.random() < pollination_probability: # 选择一个开花的植被实例 flowering_vegetation = [ (vid, vdata) for vid, vdata in vegetation_instances.items() if vdata['alive'] and vdata['type'] in ['flower'] and vdata['health'] > 0.5 ] if flowering_vegetation: target_instance_id, target_instance = random.choice(flowering_vegetation) # 提高植被的繁殖率 # 注意:这里我们只是模拟效果,实际的繁殖率调整在其他模块中处理 # 记录交互事件 self._record_interaction_event('pollination', { 'animal_species': species, 'vegetation_type': target_instance['type'], 'position': target_instance['position'] }) # 更新统计信息 self.interaction_stats['pollination_events'] += 1 except Exception as e: print(f"✗ 授粉处理失败: {e}") def _process_competition(self): """处理竞争""" try: if not self.plugin.vegetation_manager: return # 获取植被实例 vegetation_instances = self.plugin.vegetation_manager.get_all_vegetation_instances() # 按位置分组植被实例,以模拟局部竞争 position_groups = {} group_radius = 5.0 # 竞争半径 for instance_id, instance in vegetation_instances.items(): if not instance['alive']: continue pos = instance['position'] # 简化的分组方法:按坐标网格分组 grid_x = int(pos[0] // group_radius) grid_z = int(pos[2] // group_radius) group_key = (grid_x, grid_z) if group_key not in position_groups: position_groups[group_key] = [] position_groups[group_key].append((instance_id, instance)) # 处理每个组内的竞争 for group_instances in position_groups.values(): if len(group_instances) < 2: continue # 至少需要两个实例才能产生竞争 # 计算竞争强度 competition_intensity = self.interaction_params['competition_intensity'] # 对组内的每个实例应用竞争影响 for instance_id, instance in group_instances: veg_type = instance['type'] veg_info = self.plugin.vegetation_manager.get_vegetation_info(veg_type) if not veg_info: continue # 竞争因子影响 competition_factor = veg_info.get('competition_factor', 0.5) total_competition = competition_intensity * competition_factor # 根据组内其他植物数量调整竞争强度 other_plants_count = len(group_instances) - 1 adjusted_competition = total_competition * (other_plants_count / 5.0) # 标准化 # 应用竞争影响(减少健康度) health_impact = -adjusted_competition * 0.01 self.plugin.vegetation_manager.update_vegetation_health(instance_id, health_impact) # 记录交互事件 self._record_interaction_event('competition', { 'vegetation_type': veg_type, 'position': instance['position'], 'competitors_count': other_plants_count, 'health_impact': health_impact }) # 更新统计信息 self.interaction_stats['competition_events'] += 1 except Exception as e: print(f"✗ 竞争处理失败: {e}") def _process_mutualism(self): """处理互利共生""" try: if not self.plugin.animal_simulator or not self.plugin.vegetation_manager: return # 获取动物和植被信息 animal_populations = self.plugin.animal_simulator.get_population_data() vegetation_instances = self.plugin.vegetation_manager.get_all_vegetation_instances() # 定义已知的互利共生关系 mutualistic_pairs = [ ('bird', 'tree'), # 鸟类与树木 ('insect', 'flower'), # 昆虫与花卉 ] # 检查是否存在互利共生关系 for animal_species, vegetation_type in mutualistic_pairs: # 查找对应的动物种群 animal_populations_filtered = { pid: pdata for pid, pdata in animal_populations.items() if pdata['species'] == animal_species } if not animal_populations_filtered: continue # 查找对应的植被实例 vegetation_instances_filtered = { vid: vdata for vid, vdata in vegetation_instances.items() if vdata['alive'] and vdata['type'] == vegetation_type } if not vegetation_instances_filtered: continue # 计算互利共生概率 mutualism_probability = self.interaction_params['mutualism_benefit'] if random.random() < mutualism_probability: # 随机选择一个动物种群和植被实例 animal_population_id = random.choice(list(animal_populations_filtered.keys())) animal_population = animal_populations_filtered[animal_population_id] vegetation_instance_id = random.choice(list(vegetation_instances_filtered.keys())) vegetation_instance = vegetation_instances_filtered[vegetation_instance_id] # 双方都获得益处 animal_benefit = 0.01 vegetation_benefit = 0.005 # 更新动物种群健康度(这里简化处理) # 实际应用中可能需要更新个体动物状态 # 更新植被健康度 self.plugin.vegetation_manager.update_vegetation_health(vegetation_instance_id, vegetation_benefit) # 记录交互事件 self._record_interaction_event('mutualism', { 'animal_species': animal_species, 'vegetation_type': vegetation_type, 'animal_benefit': animal_benefit, 'vegetation_benefit': vegetation_benefit }) # 更新统计信息 self.interaction_stats['mutualism_events'] += 1 except Exception as e: print(f"✗ 互利共生处理失败: {e}") def _record_interaction_event(self, interaction_type: str, details: Dict[str, Any]): """ 记录交互事件 Args: interaction_type: 交互类型 details: 交互详情 """ try: event_record = { 'timestamp': time.time(), 'type': interaction_type, 'details': details } self.interaction_history.append(event_record) # 限制历史记录大小 if len(self.interaction_history) > self.max_history_size: self.interaction_history.pop(0) except Exception as e: print(f"✗ 交互事件记录失败: {e}") def get_interaction_types(self) -> Dict[str, Dict[str, Any]]: """ 获取交互类型定义 Returns: 交互类型字典 """ return self.interaction_types.copy() def set_interaction_parameter(self, parameter: str, value: float): """ 设置交互参数 Args: parameter: 参数名称 value: 参数值 """ try: if parameter in self.interaction_params: self.interaction_params[parameter] = max(0.0, min(1.0, value)) print(f"✓ 交互参数已设置: {parameter} = {value}") else: print(f"✗ 无效的交互参数: {parameter}") except Exception as e: print(f"✗ 交互参数设置失败: {e}") def get_interaction_parameters(self) -> Dict[str, float]: """ 获取交互参数 Returns: 交互参数字典 """ return self.interaction_params.copy() def get_interaction_stats(self) -> Dict[str, int]: """ 获取交互统计信息 Returns: 统计信息字典 """ return self.interaction_stats.copy() def reset_interaction_stats(self): """重置交互统计信息""" try: self.interaction_stats = { 'total_interactions': 0, 'herbivory_events': 0, 'seed_dispersal_events': 0, 'pollination_events': 0, 'competition_events': 0, 'mutualism_events': 0 } print("✓ 交互统计信息已重置") except Exception as e: print(f"✗ 交互统计信息重置失败: {e}") def get_interaction_history(self, limit: int = 50) -> List[Dict[str, Any]]: """ 获取交互历史记录 Args: limit: 返回记录数量限制 Returns: 交互历史记录列表 """ try: # 返回最近的记录 return self.interaction_history[-limit:].copy() except Exception as e: print(f"✗ 交互历史记录获取失败: {e}") return [] def clear_interaction_history(self): """清空交互历史记录""" try: self.interaction_history.clear() print("✓ 交互历史记录已清空") except Exception as e: print(f"✗ 交互历史记录清空失败: {e}") def set_time_config(self, config: Dict[str, float]): """ 设置时间配置 Args: config: 时间配置字典 """ try: self.time_config.update(config) print(f"✓ 时间配置已更新: {self.time_config}") except Exception as e: print(f"✗ 时间配置更新失败: {e}") def get_time_config(self) -> Dict[str, float]: """ 获取时间配置 Returns: 时间配置字典 """ return self.time_config.copy() def calculate_biodiversity_impact(self) -> float: """ 计算生物多样性影响 Returns: 生物多样性影响值 (-1.0 到 1.0) """ try: # 简化的生物多样性影响计算 if not self.plugin.vegetation_manager or not self.plugin.animal_simulator: return 0.0 # 获取植被和动物多样性 veg_stats = self.plugin.vegetation_manager.get_stats() animal_stats = self.plugin.animal_simulator.get_stats() veg_diversity = len([count for count in veg_stats.get('vegetation_by_type', {}).values() if count > 0]) animal_diversity = len([count for count in animal_stats.get('animals_by_species', {}).values() if count > 0]) # 简单的多样性评分 total_diversity = veg_diversity + animal_diversity max_possible_diversity = 15 # 假设最大多样性为15 diversity_score = min(1.0, total_diversity / max_possible_diversity) # 根据交互类型调整 positive_interactions = ( self.interaction_stats['seed_dispersal_events'] + self.interaction_stats['pollination_events'] + self.interaction_stats['mutualism_events'] ) negative_interactions = ( self.interaction_stats['herbivory_events'] + self.interaction_stats['competition_events'] ) interaction_balance = (positive_interactions - negative_interactions) / max(1, positive_interactions + negative_interactions) # 综合影响 overall_impact = diversity_score * 0.7 + interaction_balance * 0.3 return max(-1.0, min(1.0, overall_impact)) except Exception as e: print(f"✗ 生物多样性影响计算失败: {e}") return 0.0