EG/plugins/user/swarm_intelligence/interaction_manager.py
2025-10-30 11:46:41 +08:00

265 lines
9.9 KiB
Python

"""
群体间交互管理器
管理不同群体之间的交互行为
"""
from panda3d.core import Vec3
import random
from typing import List, Dict
class InteractionManager:
"""
群体间交互管理器
负责处理不同群体之间的交互行为,如竞争、合作、捕食等
"""
def __init__(self, config):
self.config = config
self.interactions = [] # 存储群体间交互关系
self.interaction_history = [] # 交互历史记录
def add_interaction(self, swarm1: Dict, swarm2: Dict, interaction_type: str, strength: float = 1.0):
"""
添加群体间交互关系
:param swarm1: 第一个群体
:param swarm2: 第二个群体
:param interaction_type: 交互类型 ('cooperative', 'competitive', 'predator_prey', 'neutral')
:param strength: 交互强度 (0.0-1.0)
"""
interaction = {
'swarm1': swarm1,
'swarm2': swarm2,
'type': interaction_type,
'strength': strength,
'active': True
}
self.interactions.append(interaction)
def remove_interaction(self, interaction):
"""
移除群体间交互关系
"""
if interaction in self.interactions:
self.interactions.remove(interaction)
def update_interactions(self, swarms: List[Dict]):
"""
更新所有群体间交互
:param swarms: 所有群体列表
"""
for interaction in self.interactions:
if interaction['active']:
self._process_interaction(interaction, swarms)
def _process_interaction(self, interaction: Dict, swarms: List[Dict]):
"""
处理单个群体间交互
:param interaction: 交互关系
:param swarms: 所有群体列表
"""
swarm1 = interaction['swarm1']
swarm2 = interaction['swarm2']
interaction_type = interaction['type']
strength = interaction['strength']
if interaction_type == 'cooperative':
self._process_cooperative_interaction(swarm1, swarm2, strength)
elif interaction_type == 'competitive':
self._process_competitive_interaction(swarm1, swarm2, strength)
elif interaction_type == 'predator_prey':
self._process_predator_prey_interaction(swarm1, swarm2, strength)
elif interaction_type == 'neutral':
# 中性交互,不产生影响
pass
def _process_cooperative_interaction(self, swarm1: Dict, swarm2: Dict, strength: float):
"""
处理合作交互
群体之间相互帮助,共享资源或信息
:param swarm1: 第一个群体
:param swarm2: 第二个群体
:param strength: 交互强度
"""
# 合作行为:群体成员倾向于向另一个群体的中心移动
if swarm1.members and swarm2.members:
# 计算群体中心
center1 = self._calculate_swarm_center(swarm1.members)
center2 = self._calculate_swarm_center(swarm2.members)
# 对两个群体的成员施加合作力
for member in swarm1.members:
# 向第二个群体中心移动的趋势
coop_force = (center2 - member['position']).normalized() * strength * 0.5
member['cooperative_force'] = coop_force
for member in swarm2.members:
# 向第一个群体中心移动的趋势
coop_force = (center1 - member['position']).normalized() * strength * 0.5
member['cooperative_force'] = coop_force
# 记录交互历史
self._record_interaction_history(swarm1, swarm2, 'cooperative', strength)
def _process_competitive_interaction(self, swarm1: Dict, swarm2: Dict, strength: float):
"""
处理竞争交互
群体之间争夺资源或领域
:param swarm1: 第一个群体
:param swarm2: 第二个群体
:param strength: 交互强度
"""
# 竞争行为:群体成员倾向于远离另一个群体
if swarm1.members and swarm2.members:
# 计算群体中心
center1 = self._calculate_swarm_center(swarm1.members)
center2 = self._calculate_swarm_center(swarm2.members)
# 对两个群体的成员施加竞争排斥力
for member in swarm1.members:
# 远离第二个群体中心
competition_force = (member['position'] - center2).normalized() * strength * 0.7
member['competition_force'] = competition_force
for member in swarm2.members:
# 远离第一个群体中心
competition_force = (member['position'] - center1).normalized() * strength * 0.7
member['competition_force'] = competition_force
# 记录交互历史
self._record_interaction_history(swarm1, swarm2, 'competitive', strength)
def _process_predator_prey_interaction(self, predator_swarm: Dict, prey_swarm: Dict, strength: float):
"""
处理捕食者-猎物交互
一个群体捕食另一个群体
:param predator_swarm: 捕食者群体
:param prey_swarm: 猎物群体
:param strength: 交互强度
"""
if predator_swarm.members and prey_swarm.members:
# 捕食者行为:寻找最近的猎物
for predator in predator_swarm.members:
nearest_prey = self._find_nearest_member(predator, prey_swarm.members)
if nearest_prey:
# 向猎物移动
predation_force = (nearest_prey['position'] - predator['position']).normalized() * strength
predator['predation_force'] = predation_force
# 猎物行为:逃离最近的捕食者
for prey in prey_swarm.members:
nearest_predator = self._find_nearest_member(prey, predator_swarm.members)
if nearest_predator:
# 远离捕食者
escape_force = (prey['position'] - nearest_predator['position']).normalized() * strength * 1.2
prey['escape_force'] = escape_force
# 记录交互历史
self._record_interaction_history(predator_swarm, prey_swarm, 'predator_prey', strength)
def _calculate_swarm_center(self, members: List[Dict]) -> Vec3:
"""
计算群体中心位置
:param members: 群体成员列表
:return: 群体中心位置
"""
if not members:
return Vec3(0, 0, 0)
center = Vec3(0, 0, 0)
for member in members:
center += member['position']
center /= len(members)
return center
def _find_nearest_member(self, member: Dict, other_members: List[Dict]) -> Dict:
"""
查找最近的成员
:param member: 参考成员
:param other_members: 其他成员列表
:return: 最近的成员
"""
if not other_members:
return None
nearest = None
min_distance = float('inf')
for other in other_members:
if other != member: # 排除自己
distance = (member['position'] - other['position']).length()
if distance < min_distance:
min_distance = distance
nearest = other
return nearest
def _record_interaction_history(self, swarm1: Dict, swarm2: Dict, interaction_type: str, strength: float):
"""
记录交互历史
:param swarm1: 第一个群体
:param swarm2: 第二个群体
:param interaction_type: 交互类型
:param strength: 交互强度
"""
record = {
'swarm1_id': id(swarm1),
'swarm2_id': id(swarm2),
'type': interaction_type,
'strength': strength,
'timestamp': self._get_current_time()
}
self.interaction_history.append(record)
# 限制历史记录数量
if len(self.interaction_history) > 1000:
self.interaction_history = self.interaction_history[-1000:]
def _get_current_time(self):
"""
获取当前时间戳
"""
import time
return time.time()
def get_interaction_stats(self) -> Dict:
"""
获取交互统计信息
:return: 交互统计信息
"""
stats = {
'total_interactions': len(self.interactions),
'active_interactions': len([i for i in self.interactions if i['active']]),
'cooperative': len([i for i in self.interactions if i['type'] == 'cooperative']),
'competitive': len([i for i in self.interactions if i['type'] == 'competitive']),
'predator_prey': len([i for i in self.interactions if i['type'] == 'predator_prey']),
'neutral': len([i for i in self.interactions if i['type'] == 'neutral']),
'history_count': len(self.interaction_history)
}
return stats
def clear_interactions(self):
"""
清空所有交互关系
"""
self.interactions.clear()
self.interaction_history.clear()
def get_interactions_for_swarm(self, swarm: Dict) -> List[Dict]:
"""
获取指定群体的所有交互关系
:param swarm: 群体对象
:return: 交互关系列表
"""
return [i for i in self.interactions
if i['swarm1'] == swarm or i['swarm2'] == swarm]