kards-env/kards_battle/core/battle_engine.py

647 lines
25 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
KARDS 战斗引擎核心 - 专注于战斗系统
不包含卡牌系统,单位通过函数调用直接部署
"""
from typing import Dict, List, Optional, Any, Union
from uuid import UUID
from ..battlefield.battlefield import Battlefield, HQ
from ..units.unit import Unit
from .enums import LineType
from .unit_combat_rules import UnitCombatRules
from ..events.event_system import EventType, GameEvent, publish_event
from ..events.managers.smokescreen_manager import setup_smokescreen_for_unit
class BattleEngine:
"""KARDS 战斗引擎 - 纯战斗系统"""
def __init__(self, player1_name: str, player2_name: str, debug_mode: bool = False):
self.battlefield = Battlefield(player1_name, player2_name)
self.player_names = [player1_name, player2_name]
self.debug_mode = debug_mode
self.current_turn = 1
self.active_player = 0 # 0 for player1, 1 for player2
self.turn_phase = 0 # 0=player1回合, 1=player2回合
# Kredits Slot 系统 - 指挥点槽
self.player1_kredits_slot = 0 # 当前槽数
self.player2_kredits_slot = 0
self.max_kredits_slot = 12 # 自然增长上限
self.absolute_max_kredits = 24 # 绝对上限
# Kredits 系统 - 当前回合可用指挥点
self.player1_kredits = 0 # 每回合重置为槽数
self.player2_kredits = 0
# 事件历史
self.event_history = []
# 初始化第一回合的Kredits
self._start_player_turn()
def _start_player_turn(self):
"""开始玩家回合更新Kredits Slot和Kredits"""
if self.active_player == 0:
# 玩家1在每次轮到自己时增长槽数新回合开始时
if self.turn_phase == 0 and self.player1_kredits_slot < self.max_kredits_slot:
self.player1_kredits_slot += 1
# 设置本回合Kredits为槽数
self.player1_kredits = self.player1_kredits_slot
else:
# 玩家2在每次轮到自己且完成一个完整回合后增长槽数
if self.turn_phase == 1 and self.player2_kredits_slot < self.max_kredits_slot:
self.player2_kredits_slot += 1
self.player2_kredits = self.player2_kredits_slot
def get_kredits(self, player_id: int) -> int:
"""获取玩家当前Kredits"""
if player_id == 0:
return self.player1_kredits
elif player_id == 1:
return self.player2_kredits
return 0
def get_kredits_slot(self, player_id: int) -> int:
"""获取玩家当前Kredits Slot"""
if player_id == 0:
return self.player1_kredits_slot
elif player_id == 1:
return self.player2_kredits_slot
return 0
def spend_kredits(self, player_id: int, amount: int) -> bool:
"""消耗Kredits"""
if amount <= 0:
return True
current_kredits = self.get_kredits(player_id)
if current_kredits >= amount:
if player_id == 0:
self.player1_kredits -= amount
elif player_id == 1:
self.player2_kredits -= amount
return True
return False
# === 单位直接部署功能 ===
def deploy_unit_to_support(self, unit: Unit, player_id: int, position: Optional[int] = None) -> Dict[str, Any]:
"""直接将单位部署到己方支援线的指定位置"""
if player_id not in [0, 1]:
return {"success": False, "reason": "Invalid player ID"}
support_line = self.battlefield.get_player_support_line(self.player_names[player_id])
if not support_line:
return {"success": False, "reason": "Support line not found"}
if support_line.is_full():
return {"success": False, "reason": "Support line is full"}
# 设置单位所有者
unit.owner = self.player_names[player_id]
# 如果指定了位置,尝试在该位置部署
if position is not None:
# 检查位置是否有效
if position < 0:
return {"success": False, "reason": "Invalid position: position cannot be negative"}
# 如果位置超出当前范围,添加到末尾
current_count = len(support_line.objectives)
if position >= current_count:
success = support_line.add_unit(unit)
actual_position = current_count if success else None
else:
# 在指定位置插入,其他目标向后挤
support_line.objectives.insert(position, unit)
support_line._reindex_objectives()
success = True
actual_position = position
else:
# 没有指定位置,添加到末尾
success = support_line.add_unit(unit)
objectives = support_line.get_all_objectives()
actual_position = len(objectives) - 1 if success else None
if success:
self.battlefield.unit_registry[unit.id] = unit
# 设置烟幕管理器
setup_smokescreen_for_unit(unit)
# 发布部署事件
publish_event(GameEvent(
event_type=EventType.UNIT_DEPLOYED,
source=unit,
data={
"position": unit.position,
"player_id": player_id,
"line_type": support_line.line_type,
"actual_position": actual_position
}
))
return {
"success": True,
"action": "deploy_to_support",
"unit_id": unit.id,
"player": player_id,
"position": actual_position
}
return {"success": False, "reason": "Failed to deploy unit"}
# === 单位操作(移动/攻击) ===
def move_unit(self, unit_id: UUID, target_position: tuple, player_id: int) -> Dict[str, Any]:
"""移动单位到指定位置"""
if player_id != self.active_player:
return {"success": False, "reason": "Not your turn"}
unit = self.battlefield.find_unit(unit_id)
if not unit or unit.owner != self.player_names[player_id]:
return {"success": False, "reason": "Unit not found or not yours"}
# 检查是否可以移动
if not unit.can_move():
if unit.deployed_this_turn and not unit.can_operate_immediately:
return {"success": False, "reason": "Unit was deployed this turn and cannot move"}
elif unit.has_moved_this_turn:
return {"success": False, "reason": "Unit has already moved this turn"}
elif unit.has_attacked_this_turn and not unit.can_move_and_attack:
return {"success": False, "reason": "Unit has attacked and cannot move"}
else:
return {"success": False, "reason": "Unit cannot move"}
# 检查激活成本
activation_cost = unit.stats.operation_cost
if not self.spend_kredits(player_id, activation_cost):
return {"success": False, "reason": "Insufficient Kredits for activation"}
result = self._handle_move_to_position(unit, target_position)
# 如果移动成功,标记单位已移动并发布事件
if result['success']:
old_position = unit.position
unit.perform_move()
# 发布移动事件
publish_event(GameEvent(
event_type=EventType.UNIT_MOVED,
source=unit,
data={
"old_position": old_position,
"new_position": target_position,
"player_id": player_id
}
))
# 发布位置改变事件
publish_event(GameEvent(
event_type=EventType.UNIT_POSITION_CHANGED,
source=unit,
data={
"old_position": old_position,
"new_position": unit.position
}
))
return result
def attack_target(self, attacker_id: UUID, target_id: Union[UUID, str], player_id: int) -> Dict[str, Any]:
"""攻击指定目标"""
if player_id != self.active_player:
return {"success": False, "reason": "Not your turn"}
attacker = self.battlefield.find_unit(attacker_id)
if not attacker or attacker.owner != self.player_names[player_id]:
return {"success": False, "reason": "Unit not found or not yours"}
# 检查是否可以攻击
if not attacker.can_attack():
if attacker.deployed_this_turn and not attacker.can_operate_immediately:
return {"success": False, "reason": "Unit was deployed this turn and cannot attack"}
elif attacker.attacks_this_turn >= attacker.max_attacks_per_turn:
return {"success": False, "reason": "Unit has reached maximum attacks per turn"}
elif attacker.has_moved_this_turn and not attacker.can_move_and_attack:
return {"success": False, "reason": "Unit has moved and cannot attack"}
else:
return {"success": False, "reason": "Unit cannot attack"}
# 检查激活成本
activation_cost = attacker.stats.operation_cost
if not self.spend_kredits(player_id, activation_cost):
return {"success": False, "reason": "Insufficient Kredits for activation"}
result = self._handle_attack_target(attacker, target_id)
# 如果攻击成功,标记单位已攻击并发布事件
if result['success']:
attacker.perform_attack()
# 发布单位攻击事件
publish_event(GameEvent(
event_type=EventType.UNIT_ATTACKED,
source=attacker,
data={
"target_id": target_id,
"attack_result": result
}
))
# 检查游戏是否结束HQ被摧毁
game_over, winner = self.battlefield.is_game_over()
if game_over:
result["game_over"] = True
result["winner"] = winner # winner是player_id字符串
result["winner_name"] = winner # 直接使用player_id作为名称
return result
def _handle_move_to_position(self, unit: Unit, target_pos: tuple) -> Dict[str, Any]:
"""处理移动到位置"""
target_line_type, target_index = target_pos
if not unit.position:
return {"success": False, "reason": "Unit not on battlefield"}
current_line, current_index = unit.position
# 只能从支援线移动到前线
if target_line_type != LineType.FRONT:
return {"success": False, "reason": "Can only move to front line"}
if current_line == LineType.FRONT:
return {"success": False, "reason": "Units on front line cannot move"}
return self._move_to_front_line(unit, target_index)
def _move_to_front_line(self, unit: Unit, target_index: int) -> Dict[str, Any]:
"""从支援线移动到前线"""
front_line = self.battlefield.front_line
# 检查前线控制权
if (self.battlefield.front_line_controller is not None and
self.battlefield.front_line_controller != unit.owner):
return {"success": False, "reason": "Front line controlled by enemy"}
# 检查是否要挤位置
if target_index < len(front_line.objectives):
if front_line.is_full():
return {"success": False, "reason": "Front line is full, cannot squeeze"}
else:
if front_line.is_full():
return {"success": False, "reason": "Front line is full"}
# 从支援线移除
support_line = self.battlefield.get_player_support_line(unit.owner)
if not support_line.remove_unit(unit):
return {"success": False, "reason": "Failed to remove from support line"}
# 执行移动
if target_index >= len(front_line.objectives):
front_line.add_unit(unit)
actual_position = len(front_line.objectives) - 1
else:
front_line.objectives.insert(target_index, unit)
front_line._reindex_objectives()
actual_position = target_index
# 更新前线控制权
self.battlefield._update_front_line_control()
return {
"success": True,
"action": "move_to_front",
"unit_id": unit.id,
"to_position": actual_position,
"front_line_controller": self.battlefield.front_line_controller
}
def _handle_attack_target(self, attacker: Unit, target_id: Union[UUID, str]) -> Dict[str, Any]:
"""处理攻击目标"""
# 查找目标
target = None
if isinstance(target_id, str) and target_id.startswith("hq"):
if target_id == "hq1":
target = self.battlefield.player1_hq
elif target_id == "hq2":
target = self.battlefield.player2_hq
else:
target = self.battlefield.find_unit(target_id)
if not target:
return {"success": False, "reason": "Target not found"}
# 检查是否能攻击该目标
if not UnitCombatRules.can_attack_target(attacker, target, self.battlefield):
return {"success": False, "reason": "Cannot attack this target"}
# 发布攻击前检查事件(可取消)
before_attack_event = GameEvent(
event_type=EventType.BEFORE_ATTACK_CHECK,
source=attacker,
target=target,
data={"attack_type": "unit_attack"},
cancellable=True
)
publish_event(before_attack_event)
# 如果攻击被阻止,返回失败
if before_attack_event.cancelled:
return {"success": False, "reason": before_attack_event.cancel_reason}
# 执行攻击
return self._execute_attack(attacker, target)
def _execute_attack(self, attacker: Unit, target: Union[Unit, HQ]) -> Dict[str, Any]:
"""执行攻击"""
damage = attacker.get_effective_attack()
if isinstance(target, Unit):
return self._unit_combat(attacker, target)
else:
# 攻击HQ
actual_damage = target.take_damage(damage)
return {
"success": True,
"action": "attack_hq",
"attacker_id": attacker.id,
"target": "HQ",
"damage_dealt": actual_damage,
"hq_destroyed": target.is_destroyed()
}
def _unit_combat(self, attacker: Unit, target: Unit) -> Dict[str, Any]:
"""单位间战斗"""
# 处理伏击
if target.has_keyword("AMBUSH"):
ambush_damage = target.get_effective_attack()
attacker_damage = attacker.stats.take_damage(ambush_damage)
if not attacker.stats.is_alive():
self.battlefield.remove_unit(attacker)
return {
"success": True,
"action": "attack_unit",
"ambush_triggered": True,
"attacker_destroyed": True,
"ambush_damage": attacker_damage
}
# 计算攻击伤害
attack_damage = attacker.get_effective_attack()
# 应用重甲减伤
for keyword in target.keywords:
if keyword.startswith("HEAVY_ARMOR_"):
armor_value = int(keyword.split("_")[-1])
attack_damage = max(0, attack_damage - armor_value)
break
# 战斗是同时进行的:先计算双方攻击力,再同时造成伤害
# 1. 计算双方攻击力
counter_attack = 0
counter_damage = 0
# 检查防御方是否能够反击
if UnitCombatRules.can_counter_attack(target, attacker):
counter_attack = target.get_effective_attack()
# 2. 同时造成伤害
actual_damage = target.stats.take_damage(attack_damage)
if counter_attack > 0:
counter_damage = attacker.stats.take_damage(counter_attack)
# 3. 发布伤害事件
if actual_damage > 0:
publish_event(GameEvent(
event_type=EventType.UNIT_TAKES_DAMAGE,
source=attacker,
target=target,
data={
"damage": actual_damage,
"damage_type": "attack"
}
))
if counter_damage > 0:
publish_event(GameEvent(
event_type=EventType.UNIT_COUNTER_ATTACKS,
source=target,
target=attacker,
data={
"damage": counter_damage,
"counter_attack_damage": counter_attack
}
))
publish_event(GameEvent(
event_type=EventType.UNIT_TAKES_DAMAGE,
source=target,
target=attacker,
data={
"damage": counter_damage,
"damage_type": "counter_attack"
}
))
# 检查单位死亡并发布事件
units_destroyed = []
if not target.stats.is_alive():
self.battlefield.remove_unit(target)
units_destroyed.append(target.id)
# 发布单位摧毁事件
publish_event(GameEvent(
event_type=EventType.UNIT_DESTROYED,
source=target,
data={
"destroyed_by": attacker.id,
"destruction_cause": "combat_damage"
}
))
if not attacker.stats.is_alive():
self.battlefield.remove_unit(attacker)
units_destroyed.append(attacker.id)
# 发布单位摧毁事件
publish_event(GameEvent(
event_type=EventType.UNIT_DESTROYED,
source=attacker,
data={
"destroyed_by": target.id,
"destruction_cause": "counter_attack_damage"
}
))
return {
"success": True,
"action": "attack_unit",
"attacker_id": attacker.id,
"target_id": target.id,
"damage_dealt": actual_damage,
"counter_damage": counter_damage,
"units_destroyed": units_destroyed
}
# === 回合管理 ===
def end_turn(self) -> Dict[str, Any]:
"""结束当前玩家的回合"""
old_player = self.active_player
old_phase = self.turn_phase
# 切换玩家
if self.turn_phase == 0:
# 从玩家1切换到玩家2但还在同一回合
self.active_player = 1
self.turn_phase = 1
else:
# 从玩家2切换回玩家1进入下一回合
self.active_player = 0
self.turn_phase = 0
self.current_turn += 1 # 只有双方都行动过才算一个完整回合
# 开始新玩家回合更新Kredits Slot和Kredits
self._start_player_turn()
# 重置活跃玩家的单位状态
for unit in self.battlefield.get_all_units(self.player_names[self.active_player]):
unit.start_new_turn() # 使用新的方法重置所有状态
# 检查游戏结束
game_over, winner = self.battlefield.is_game_over()
return {
"turn_ended": True,
"previous_player": old_player,
"new_active_player": self.active_player,
"turn_number": self.current_turn,
"turn_phase": self.turn_phase,
"is_new_round": self.turn_phase == 0, # 是否是新一轮的开始
"kredits": self.get_kredits(self.active_player),
"kredits_slot": self.get_kredits_slot(self.active_player),
"game_over": game_over,
"winner": winner
}
# === DEBUG 功能 ===
def debug_set_unit_stats(self, unit_id: UUID, attack: Optional[int] = None,
defense: Optional[int] = None) -> Dict[str, Any]:
"""DEBUG: 设置单位属性"""
if not self.debug_mode:
return {"success": False, "reason": "Debug mode not enabled"}
unit = self.battlefield.find_unit(unit_id)
if not unit:
return {"success": False, "reason": "Unit not found"}
changes = {}
if attack is not None:
unit.stats.attack = attack
changes["attack"] = attack
if defense is not None:
unit.stats.current_defense = defense
unit.stats.max_defense = defense
changes["defense"] = defense
return {
"success": True,
"action": "debug_set_unit_stats",
"unit_id": unit_id,
"changes": changes
}
def debug_set_kredits(self, player_id: int, kredits: Optional[int] = None,
kredits_slot: Optional[int] = None) -> Dict[str, Any]:
"""DEBUG: 设置玩家Kredits和Kredits Slot"""
if not self.debug_mode:
return {"success": False, "reason": "Debug mode not enabled"}
if player_id not in [0, 1]:
return {"success": False, "reason": "Invalid player ID"}
changes = {}
if kredits is not None:
kredits = max(0, min(kredits, self.absolute_max_kredits))
if player_id == 0:
self.player1_kredits = kredits
else:
self.player2_kredits = kredits
changes["kredits"] = kredits
if kredits_slot is not None:
kredits_slot = max(0, min(kredits_slot, self.absolute_max_kredits))
if player_id == 0:
self.player1_kredits_slot = kredits_slot
else:
self.player2_kredits_slot = kredits_slot
changes["kredits_slot"] = kredits_slot
return {
"success": True,
"action": "debug_set_kredits",
"player_id": player_id,
"changes": changes
}
def debug_set_hq_defense(self, player_id: int, defense: int) -> Dict[str, Any]:
"""DEBUG: 设置HQ防御值"""
if not self.debug_mode:
return {"success": False, "reason": "Debug mode not enabled"}
hq = None
if player_id == 0:
hq = self.battlefield.player1_hq
elif player_id == 1:
hq = self.battlefield.player2_hq
else:
return {"success": False, "reason": "Invalid player ID"}
defense = max(0, defense)
hq.current_defense = defense
# HQ没有max_defense属性只更新current_defense
return {
"success": True,
"action": "debug_set_hq_defense",
"player_id": player_id,
"new_defense": defense
}
# === 游戏状态 ===
def get_game_state(self) -> Dict[str, Any]:
"""获取完整游戏状态"""
return {
"turn": self.current_turn,
"turn_phase": self.turn_phase,
"active_player": self.active_player,
"kredits": {
"player1": self.player1_kredits,
"player2": self.player2_kredits,
"active_player": self.get_kredits(self.active_player)
},
"kredits_slots": {
"player1": self.player1_kredits_slot,
"player2": self.player2_kredits_slot,
"active_player": self.get_kredits_slot(self.active_player)
},
"battlefield": str(self.battlefield),
"player1_hq": self.battlefield.player1_hq.current_defense,
"player2_hq": self.battlefield.player2_hq.current_defense,
"front_line_controller": self.battlefield.front_line_controller,
"units_count": {
"player1_total": len(self.battlefield.get_all_units(self.player_names[0])),
"player2_total": len(self.battlefield.get_all_units(self.player_names[1])),
"front_line": len(self.battlefield.front_line.get_all_units())
},
"debug_mode": self.debug_mode
}