365 lines
13 KiB
Python
365 lines
13 KiB
Python
|
"""
|
||
|
|
游戏引擎核心
|
||
|
|
处理游戏逻辑、事件触发和状态管理
|
||
|
|
"""
|
||
|
|
from typing import Dict, List, Optional, Any
|
||
|
|
from uuid import UUID
|
||
|
|
|
||
|
|
from ..battlefield.battlefield import Battlefield
|
||
|
|
from ..units.unit import Unit
|
||
|
|
from ..abilities.keywords import TriggerTiming, KeywordRegistry
|
||
|
|
from ..abilities.abilities import AbilityManager
|
||
|
|
from .enums import GamePhase
|
||
|
|
|
||
|
|
|
||
|
|
class GameEvent:
|
||
|
|
"""游戏事件"""
|
||
|
|
|
||
|
|
def __init__(self, event_type: str, source: Optional[Unit] = None,
|
||
|
|
target: Optional[Unit] = None, context: Dict[str, Any] = None):
|
||
|
|
self.event_type = event_type
|
||
|
|
self.source = source
|
||
|
|
self.target = target
|
||
|
|
self.context = context or {}
|
||
|
|
self.timestamp = 0 # 由事件管理器设置
|
||
|
|
|
||
|
|
|
||
|
|
class EventManager:
|
||
|
|
"""事件管理器"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.event_queue = []
|
||
|
|
self.event_listeners = {}
|
||
|
|
self.event_counter = 0
|
||
|
|
|
||
|
|
def add_event(self, event: GameEvent) -> None:
|
||
|
|
"""添加事件到队列"""
|
||
|
|
event.timestamp = self.event_counter
|
||
|
|
self.event_counter += 1
|
||
|
|
self.event_queue.append(event)
|
||
|
|
|
||
|
|
def process_events(self, battlefield: Battlefield) -> List[Dict[str, Any]]:
|
||
|
|
"""处理所有事件"""
|
||
|
|
results = []
|
||
|
|
|
||
|
|
while self.event_queue:
|
||
|
|
event = self.event_queue.pop(0)
|
||
|
|
result = self._process_single_event(event, battlefield)
|
||
|
|
if result:
|
||
|
|
results.append(result)
|
||
|
|
|
||
|
|
return results
|
||
|
|
|
||
|
|
def _process_single_event(self, event: GameEvent, battlefield: Battlefield) -> Optional[Dict[str, Any]]:
|
||
|
|
"""处理单个事件"""
|
||
|
|
# 根据事件类型触发相应的能力
|
||
|
|
timing_map = {
|
||
|
|
"unit_deployed": TriggerTiming.ON_DEPLOY,
|
||
|
|
"unit_attacks": TriggerTiming.ON_ATTACK,
|
||
|
|
"unit_defends": TriggerTiming.ON_DEFEND,
|
||
|
|
"unit_takes_damage": TriggerTiming.ON_DAMAGE_TAKEN,
|
||
|
|
"unit_dies": TriggerTiming.ON_DEATH,
|
||
|
|
"turn_start": TriggerTiming.ON_TURN_START,
|
||
|
|
"turn_end": TriggerTiming.ON_TURN_END,
|
||
|
|
"before_combat": TriggerTiming.BEFORE_COMBAT,
|
||
|
|
"after_combat": TriggerTiming.AFTER_COMBAT
|
||
|
|
}
|
||
|
|
|
||
|
|
trigger_timing = timing_map.get(event.event_type)
|
||
|
|
if not trigger_timing:
|
||
|
|
return None
|
||
|
|
|
||
|
|
# 触发所有相关单位的能力
|
||
|
|
all_units = battlefield.get_all_units()
|
||
|
|
ability_manager = AbilityManager()
|
||
|
|
|
||
|
|
results = []
|
||
|
|
for unit in all_units:
|
||
|
|
# 触发关键词效果
|
||
|
|
self._trigger_keyword_effects(unit, battlefield, trigger_timing, event.context)
|
||
|
|
|
||
|
|
# 触发能力
|
||
|
|
unit_results = ability_manager.trigger_abilities(unit, battlefield, trigger_timing, event.context)
|
||
|
|
results.extend(unit_results)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"event": event.event_type,
|
||
|
|
"results": results
|
||
|
|
} if results else None
|
||
|
|
|
||
|
|
def _trigger_keyword_effects(self, unit: Unit, battlefield: Battlefield,
|
||
|
|
timing: TriggerTiming, context: Dict[str, Any]) -> None:
|
||
|
|
"""触发单位的关键词效果"""
|
||
|
|
for keyword_name in unit.keywords:
|
||
|
|
keyword_effect = KeywordRegistry.create_keyword(keyword_name)
|
||
|
|
if keyword_effect:
|
||
|
|
keyword_effect.apply_effect(unit, battlefield, timing, context)
|
||
|
|
|
||
|
|
|
||
|
|
class CombatResolver:
|
||
|
|
"""战斗解算器"""
|
||
|
|
|
||
|
|
def __init__(self, event_manager: EventManager):
|
||
|
|
self.event_manager = event_manager
|
||
|
|
|
||
|
|
def resolve_combat(self, attacker: Unit, target: Unit, battlefield: Battlefield) -> Dict[str, Any]:
|
||
|
|
"""解算战斗"""
|
||
|
|
# 战斗前事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"before_combat",
|
||
|
|
source=attacker,
|
||
|
|
target=target,
|
||
|
|
context={"combat_type": "attack"}
|
||
|
|
))
|
||
|
|
|
||
|
|
# 处理伏击
|
||
|
|
if target.has_keyword("AMBUSH"):
|
||
|
|
ambush_result = self._resolve_ambush(target, attacker, battlefield)
|
||
|
|
if ambush_result.get("attacker_destroyed"):
|
||
|
|
return ambush_result
|
||
|
|
|
||
|
|
# 计算最终伤害
|
||
|
|
base_damage = attacker.get_effective_attack()
|
||
|
|
final_damage = self._calculate_damage(base_damage, target)
|
||
|
|
|
||
|
|
# 应用伤害
|
||
|
|
actual_damage = target.stats.take_damage(final_damage)
|
||
|
|
|
||
|
|
# 触发受伤事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"unit_takes_damage",
|
||
|
|
source=attacker,
|
||
|
|
target=target,
|
||
|
|
context={"damage": actual_damage}
|
||
|
|
))
|
||
|
|
|
||
|
|
# 反击（如果目标存活且有反击能力）
|
||
|
|
counter_damage = 0
|
||
|
|
if target.stats.is_alive() and self._can_counter_attack(target, attacker):
|
||
|
|
counter_base_damage = target.get_effective_attack()
|
||
|
|
counter_final_damage = self._calculate_damage(counter_base_damage, attacker)
|
||
|
|
counter_damage = attacker.stats.take_damage(counter_final_damage)
|
||
|
|
|
||
|
|
if counter_damage > 0:
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"unit_takes_damage",
|
||
|
|
source=target,
|
||
|
|
target=attacker,
|
||
|
|
context={"damage": counter_damage, "is_counter": True}
|
||
|
|
))
|
||
|
|
|
||
|
|
# 检查单位死亡
|
||
|
|
units_destroyed = []
|
||
|
|
if not target.stats.is_alive():
|
||
|
|
self.event_manager.add_event(GameEvent("unit_dies", target=target))
|
||
|
|
battlefield.remove_unit(target)
|
||
|
|
units_destroyed.append(target.id)
|
||
|
|
|
||
|
|
if not attacker.stats.is_alive():
|
||
|
|
self.event_manager.add_event(GameEvent("unit_dies", target=attacker))
|
||
|
|
battlefield.remove_unit(attacker)
|
||
|
|
units_destroyed.append(attacker.id)
|
||
|
|
|
||
|
|
# 战斗后事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"after_combat",
|
||
|
|
source=attacker,
|
||
|
|
target=target,
|
||
|
|
context={"damage_dealt": actual_damage, "counter_damage": counter_damage}
|
||
|
|
))
|
||
|
|
|
||
|
|
return {
|
||
|
|
"attacker_id": attacker.id,
|
||
|
|
"target_id": target.id,
|
||
|
|
"damage_dealt": actual_damage,
|
||
|
|
"counter_damage": counter_damage,
|
||
|
|
"units_destroyed": units_destroyed,
|
||
|
|
"combat_resolved": True
|
||
|
|
}
|
||
|
|
|
||
|
|
def _resolve_ambush(self, defender: Unit, attacker: Unit, battlefield: Battlefield) -> Dict[str, Any]:
|
||
|
|
"""解算伏击"""
|
||
|
|
ambush_damage = defender.get_effective_attack()
|
||
|
|
actual_damage = attacker.stats.take_damage(ambush_damage)
|
||
|
|
|
||
|
|
result = {
|
||
|
|
"ambush_triggered": True,
|
||
|
|
"ambush_damage": actual_damage,
|
||
|
|
"attacker_destroyed": not attacker.stats.is_alive()
|
||
|
|
}
|
||
|
|
|
||
|
|
if not attacker.stats.is_alive():
|
||
|
|
self.event_manager.add_event(GameEvent("unit_dies", target=attacker))
|
||
|
|
battlefield.remove_unit(attacker)
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
def _calculate_damage(self, base_damage: int, target: Unit) -> int:
|
||
|
|
"""计算最终伤害（考虑重甲等）"""
|
||
|
|
final_damage = base_damage
|
||
|
|
|
||
|
|
# 重甲减伤
|
||
|
|
for keyword in target.keywords:
|
||
|
|
if keyword.startswith("HEAVY_ARMOR_"):
|
||
|
|
try:
|
||
|
|
armor_value = int(keyword.split("_")[-1])
|
||
|
|
final_damage = max(0, final_damage - armor_value)
|
||
|
|
break
|
||
|
|
except ValueError:
|
||
|
|
continue
|
||
|
|
|
||
|
|
return final_damage
|
||
|
|
|
||
|
|
def _can_counter_attack(self, defender: Unit, attacker: Unit) -> bool:
|
||
|
|
"""检查是否可以反击"""
|
||
|
|
# 轰炸机攻击时目标不能反击（除非目标也是战斗机）
|
||
|
|
from ..core.enums import UnitType
|
||
|
|
if attacker.unit_type == UnitType.BOMBER and defender.unit_type != UnitType.FIGHTER:
|
||
|
|
return False
|
||
|
|
|
||
|
|
# 被压制的单位不能反击
|
||
|
|
if defender.has_keyword("PINNED"):
|
||
|
|
return False
|
||
|
|
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
class GameEngine:
|
||
|
|
"""游戏引擎主类"""
|
||
|
|
|
||
|
|
def __init__(self, player1_id: str, player2_id: str):
|
||
|
|
self.battlefield = Battlefield(player1_id, player2_id)
|
||
|
|
self.event_manager = EventManager()
|
||
|
|
self.combat_resolver = CombatResolver(self.event_manager)
|
||
|
|
|
||
|
|
self.current_turn = 1
|
||
|
|
self.active_player = player1_id
|
||
|
|
self.game_phase = GamePhase.DEPLOY
|
||
|
|
|
||
|
|
self.turn_history = []
|
||
|
|
|
||
|
|
def deploy_unit(self, unit: Unit, player_id: str, line_type, position: int) -> Dict[str, Any]:
|
||
|
|
"""部署单位"""
|
||
|
|
if self.game_phase != GamePhase.DEPLOY:
|
||
|
|
return {"success": False, "reason": "Not in deploy phase"}
|
||
|
|
|
||
|
|
if player_id != self.active_player:
|
||
|
|
return {"success": False, "reason": "Not your turn"}
|
||
|
|
|
||
|
|
success = self.battlefield.deploy_unit(unit, player_id, line_type, position)
|
||
|
|
if success:
|
||
|
|
# 触发部署事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"unit_deployed",
|
||
|
|
source=unit,
|
||
|
|
context={"line_type": line_type, "position": position}
|
||
|
|
))
|
||
|
|
|
||
|
|
# 处理事件
|
||
|
|
self.event_manager.process_events(self.battlefield)
|
||
|
|
|
||
|
|
return {"success": True, "unit_deployed": unit.id}
|
||
|
|
|
||
|
|
return {"success": False, "reason": "Failed to deploy unit"}
|
||
|
|
|
||
|
|
def attack_unit(self, attacker_id: UUID, target_id: UUID) -> Dict[str, Any]:
|
||
|
|
"""单位攻击"""
|
||
|
|
if self.game_phase != GamePhase.OPERATION:
|
||
|
|
return {"success": False, "reason": "Not in operation phase"}
|
||
|
|
|
||
|
|
attacker = self.battlefield.find_unit(attacker_id)
|
||
|
|
target = self.battlefield.find_unit(target_id)
|
||
|
|
|
||
|
|
if not attacker or not target:
|
||
|
|
return {"success": False, "reason": "Unit not found"}
|
||
|
|
|
||
|
|
if attacker.owner != self.active_player:
|
||
|
|
return {"success": False, "reason": "Not your unit"}
|
||
|
|
|
||
|
|
if not attacker.can_attack_target(target, self.battlefield):
|
||
|
|
return {"success": False, "reason": "Cannot attack target"}
|
||
|
|
|
||
|
|
if not target.can_be_attacked_by(attacker, self.battlefield):
|
||
|
|
return {"success": False, "reason": "Target cannot be attacked"}
|
||
|
|
|
||
|
|
# 检查单位是否已经行动过
|
||
|
|
if attacker.has_operated_this_turn and not attacker.has_keyword("FURY"):
|
||
|
|
return {"success": False, "reason": "Unit already operated this turn"}
|
||
|
|
|
||
|
|
# 触发攻击事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"unit_attacks",
|
||
|
|
source=attacker,
|
||
|
|
target=target
|
||
|
|
))
|
||
|
|
|
||
|
|
# 解算战斗
|
||
|
|
combat_result = self.combat_resolver.resolve_combat(attacker, target, self.battlefield)
|
||
|
|
|
||
|
|
# 标记单位已行动
|
||
|
|
attacker.has_operated_this_turn = True
|
||
|
|
|
||
|
|
# 处理所有事件
|
||
|
|
event_results = self.event_manager.process_events(self.battlefield)
|
||
|
|
|
||
|
|
return {
|
||
|
|
**combat_result,
|
||
|
|
"event_results": event_results
|
||
|
|
}
|
||
|
|
|
||
|
|
def end_turn(self) -> Dict[str, Any]:
|
||
|
|
"""结束回合"""
|
||
|
|
# 触发回合结束事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"turn_end",
|
||
|
|
context={"player": self.active_player, "turn": self.current_turn}
|
||
|
|
))
|
||
|
|
|
||
|
|
# 处理事件
|
||
|
|
self.event_manager.process_events(self.battlefield)
|
||
|
|
|
||
|
|
# 切换玩家
|
||
|
|
self.active_player = (self.battlefield.player2_id if self.active_player == self.battlefield.player1_id
|
||
|
|
else self.battlefield.player1_id)
|
||
|
|
|
||
|
|
# 重置单位状态
|
||
|
|
for unit in self.battlefield.get_all_units(self.active_player):
|
||
|
|
unit.reset_operation_status()
|
||
|
|
|
||
|
|
# 触发回合开始事件
|
||
|
|
self.event_manager.add_event(GameEvent(
|
||
|
|
"turn_start",
|
||
|
|
context={"player": self.active_player, "turn": self.current_turn}
|
||
|
|
))
|
||
|
|
|
||
|
|
# 处理事件
|
||
|
|
self.event_manager.process_events(self.battlefield)
|
||
|
|
|
||
|
|
self.current_turn += 1
|
||
|
|
self.game_phase = GamePhase.DEPLOY
|
||
|
|
|
||
|
|
# 检查游戏是否结束
|
||
|
|
game_over, winner = self.battlefield.is_game_over()
|
||
|
|
|
||
|
|
return {
|
||
|
|
"turn_ended": True,
|
||
|
|
"new_active_player": self.active_player,
|
||
|
|
"turn_number": self.current_turn,
|
||
|
|
"game_over": game_over,
|
||
|
|
"winner": winner
|
||
|
|
}
|
||
|
|
|
||
|
|
def get_game_state(self) -> Dict[str, Any]:
|
||
|
|
"""获取游戏状态"""
|
||
|
|
return {
|
||
|
|
"turn": self.current_turn,
|
||
|
|
"active_player": self.active_player,
|
||
|
|
"phase": self.game_phase.value,
|
||
|
|
"battlefield": str(self.battlefield),
|
||
|
|
"player1_hq": self.battlefield.player1_hq.current_defense,
|
||
|
|
"player2_hq": self.battlefield.player2_hq.current_defense,
|
||
|
|
"units_count": {
|
||
|
|
"player1": len(self.battlefield.get_all_units(self.battlefield.player1_id)),
|
||
|
|
"player2": len(self.battlefield.get_all_units(self.battlefield.player2_id))
|
||
|
|
}
|
||
|
|
}
|