470 lines
16 KiB
Python
470 lines
16 KiB
Python
"""
|
|
Combat system data models.
|
|
|
|
This module defines the combat-related dataclasses including Combatant (a wrapper
|
|
for characters/enemies in combat) and CombatEncounter (the combat state manager).
|
|
"""
|
|
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Dict, Any, List, Optional
|
|
import random
|
|
|
|
from app.models.stats import Stats
|
|
from app.models.effects import Effect
|
|
from app.models.abilities import Ability
|
|
from app.models.enums import CombatStatus, EffectType, DamageType
|
|
|
|
|
|
@dataclass
|
|
class Combatant:
|
|
"""
|
|
Represents a character or enemy in combat.
|
|
|
|
This wraps either a player Character or an NPC/enemy for combat purposes,
|
|
tracking combat-specific state like current HP/MP, active effects, and cooldowns.
|
|
|
|
Attributes:
|
|
combatant_id: Unique identifier (character_id or enemy_id)
|
|
name: Display name
|
|
is_player: True if player character, False if NPC/enemy
|
|
current_hp: Current hit points
|
|
max_hp: Maximum hit points
|
|
current_mp: Current mana points
|
|
max_mp: Maximum mana points
|
|
stats: Current combat stats (use get_effective_stats() from Character)
|
|
active_effects: Effects currently applied to this combatant
|
|
abilities: Available abilities for this combatant
|
|
cooldowns: Map of ability_id to turns remaining
|
|
initiative: Turn order value (rolled at combat start)
|
|
weapon_crit_chance: Critical hit chance from equipped weapon
|
|
weapon_crit_multiplier: Critical hit damage multiplier
|
|
weapon_damage_type: Primary damage type of weapon
|
|
elemental_damage_type: Secondary damage type for elemental weapons
|
|
physical_ratio: Portion of damage that is physical (0.0-1.0)
|
|
elemental_ratio: Portion of damage that is elemental (0.0-1.0)
|
|
"""
|
|
|
|
combatant_id: str
|
|
name: str
|
|
is_player: bool
|
|
current_hp: int
|
|
max_hp: int
|
|
current_mp: int
|
|
max_mp: int
|
|
stats: Stats
|
|
active_effects: List[Effect] = field(default_factory=list)
|
|
abilities: List[str] = field(default_factory=list) # ability_ids
|
|
cooldowns: Dict[str, int] = field(default_factory=dict)
|
|
initiative: int = 0
|
|
|
|
# Weapon properties (for combat calculations)
|
|
weapon_crit_chance: float = 0.05
|
|
weapon_crit_multiplier: float = 2.0
|
|
weapon_damage_type: Optional[DamageType] = None
|
|
|
|
# Elemental weapon properties (for split damage)
|
|
elemental_damage_type: Optional[DamageType] = None
|
|
physical_ratio: float = 1.0
|
|
elemental_ratio: float = 0.0
|
|
|
|
def is_alive(self) -> bool:
|
|
"""Check if combatant is still alive."""
|
|
return self.current_hp > 0
|
|
|
|
def is_dead(self) -> bool:
|
|
"""Check if combatant is dead."""
|
|
return self.current_hp <= 0
|
|
|
|
def is_stunned(self) -> bool:
|
|
"""Check if combatant is stunned and cannot act."""
|
|
return any(e.effect_type == EffectType.STUN for e in self.active_effects)
|
|
|
|
def take_damage(self, damage: int) -> int:
|
|
"""
|
|
Apply damage to this combatant.
|
|
|
|
Damage is reduced by shields first, then HP.
|
|
|
|
Args:
|
|
damage: Amount of damage to apply
|
|
|
|
Returns:
|
|
Actual damage dealt to HP (after shields)
|
|
"""
|
|
remaining_damage = damage
|
|
|
|
# Apply shield absorption
|
|
for effect in self.active_effects:
|
|
if effect.effect_type == EffectType.SHIELD and remaining_damage > 0:
|
|
remaining_damage = effect.reduce_shield(remaining_damage)
|
|
|
|
# Apply remaining damage to HP
|
|
hp_damage = min(remaining_damage, self.current_hp)
|
|
self.current_hp -= hp_damage
|
|
|
|
return hp_damage
|
|
|
|
def heal(self, amount: int) -> int:
|
|
"""
|
|
Heal this combatant.
|
|
|
|
Args:
|
|
amount: Amount to heal
|
|
|
|
Returns:
|
|
Actual amount healed (capped at max_hp)
|
|
"""
|
|
old_hp = self.current_hp
|
|
self.current_hp = min(self.max_hp, self.current_hp + amount)
|
|
return self.current_hp - old_hp
|
|
|
|
def restore_mana(self, amount: int) -> int:
|
|
"""
|
|
Restore mana to this combatant.
|
|
|
|
Args:
|
|
amount: Amount to restore
|
|
|
|
Returns:
|
|
Actual amount restored (capped at max_mp)
|
|
"""
|
|
old_mp = self.current_mp
|
|
self.current_mp = min(self.max_mp, self.current_mp + amount)
|
|
return self.current_mp - old_mp
|
|
|
|
def can_use_ability(self, ability_id: str, ability: Ability) -> bool:
|
|
"""
|
|
Check if ability can be used right now.
|
|
|
|
Args:
|
|
ability_id: Ability identifier
|
|
ability: Ability instance
|
|
|
|
Returns:
|
|
True if ability can be used, False otherwise
|
|
"""
|
|
# Check if ability is available to this combatant
|
|
if ability_id not in self.abilities:
|
|
return False
|
|
|
|
# Check mana cost
|
|
if self.current_mp < ability.mana_cost:
|
|
return False
|
|
|
|
# Check cooldown
|
|
if ability_id in self.cooldowns and self.cooldowns[ability_id] > 0:
|
|
return False
|
|
|
|
return True
|
|
|
|
def use_ability_cost(self, ability: Ability, ability_id: str) -> None:
|
|
"""
|
|
Apply the costs of using an ability (mana, cooldown).
|
|
|
|
Args:
|
|
ability: Ability being used
|
|
ability_id: Ability identifier
|
|
"""
|
|
# Consume mana
|
|
self.current_mp -= ability.mana_cost
|
|
|
|
# Set cooldown
|
|
if ability.cooldown > 0:
|
|
self.cooldowns[ability_id] = ability.cooldown
|
|
|
|
def tick_effects(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
Process all active effects for this turn.
|
|
|
|
Returns:
|
|
List of effect tick results
|
|
"""
|
|
results = []
|
|
expired_effects = []
|
|
|
|
for effect in self.active_effects:
|
|
result = effect.tick()
|
|
|
|
# Apply effect results
|
|
if effect.effect_type == EffectType.DOT:
|
|
self.take_damage(result["value"])
|
|
elif effect.effect_type == EffectType.HOT:
|
|
self.heal(result["value"])
|
|
|
|
results.append(result)
|
|
|
|
# Mark expired effects for removal
|
|
if result.get("expired", False):
|
|
expired_effects.append(effect)
|
|
|
|
# Remove expired effects
|
|
for effect in expired_effects:
|
|
self.active_effects.remove(effect)
|
|
|
|
return results
|
|
|
|
def tick_cooldowns(self) -> None:
|
|
"""Reduce all ability cooldowns by 1 turn."""
|
|
for ability_id in list(self.cooldowns.keys()):
|
|
self.cooldowns[ability_id] -= 1
|
|
if self.cooldowns[ability_id] <= 0:
|
|
del self.cooldowns[ability_id]
|
|
|
|
def add_effect(self, effect: Effect) -> None:
|
|
"""
|
|
Add an effect to this combatant.
|
|
|
|
If the same effect already exists, stack it instead.
|
|
|
|
Args:
|
|
effect: Effect to add
|
|
"""
|
|
# Check if effect already exists
|
|
for existing in self.active_effects:
|
|
if existing.name == effect.name and existing.effect_type == effect.effect_type:
|
|
# Stack the effect
|
|
existing.apply_stack(effect.duration)
|
|
return
|
|
|
|
# New effect, add it
|
|
self.active_effects.append(effect)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize combatant to dictionary."""
|
|
return {
|
|
"combatant_id": self.combatant_id,
|
|
"name": self.name,
|
|
"is_player": self.is_player,
|
|
"current_hp": self.current_hp,
|
|
"max_hp": self.max_hp,
|
|
"current_mp": self.current_mp,
|
|
"max_mp": self.max_mp,
|
|
"stats": self.stats.to_dict(),
|
|
"active_effects": [e.to_dict() for e in self.active_effects],
|
|
"abilities": self.abilities,
|
|
"cooldowns": self.cooldowns,
|
|
"initiative": self.initiative,
|
|
"weapon_crit_chance": self.weapon_crit_chance,
|
|
"weapon_crit_multiplier": self.weapon_crit_multiplier,
|
|
"weapon_damage_type": self.weapon_damage_type.value if self.weapon_damage_type else None,
|
|
"elemental_damage_type": self.elemental_damage_type.value if self.elemental_damage_type else None,
|
|
"physical_ratio": self.physical_ratio,
|
|
"elemental_ratio": self.elemental_ratio,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'Combatant':
|
|
"""Deserialize combatant from dictionary."""
|
|
stats = Stats.from_dict(data["stats"])
|
|
active_effects = [Effect.from_dict(e) for e in data.get("active_effects", [])]
|
|
|
|
# Parse damage types
|
|
weapon_damage_type = None
|
|
if data.get("weapon_damage_type"):
|
|
weapon_damage_type = DamageType(data["weapon_damage_type"])
|
|
|
|
elemental_damage_type = None
|
|
if data.get("elemental_damage_type"):
|
|
elemental_damage_type = DamageType(data["elemental_damage_type"])
|
|
|
|
return cls(
|
|
combatant_id=data["combatant_id"],
|
|
name=data["name"],
|
|
is_player=data["is_player"],
|
|
current_hp=data["current_hp"],
|
|
max_hp=data["max_hp"],
|
|
current_mp=data["current_mp"],
|
|
max_mp=data["max_mp"],
|
|
stats=stats,
|
|
active_effects=active_effects,
|
|
abilities=data.get("abilities", []),
|
|
cooldowns=data.get("cooldowns", {}),
|
|
initiative=data.get("initiative", 0),
|
|
weapon_crit_chance=data.get("weapon_crit_chance", 0.05),
|
|
weapon_crit_multiplier=data.get("weapon_crit_multiplier", 2.0),
|
|
weapon_damage_type=weapon_damage_type,
|
|
elemental_damage_type=elemental_damage_type,
|
|
physical_ratio=data.get("physical_ratio", 1.0),
|
|
elemental_ratio=data.get("elemental_ratio", 0.0),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class CombatEncounter:
|
|
"""
|
|
Represents a combat encounter state.
|
|
|
|
Manages turn order, combatants, combat log, and victory/defeat conditions.
|
|
|
|
Attributes:
|
|
encounter_id: Unique identifier
|
|
combatants: All fighters in this combat
|
|
turn_order: Combatant IDs sorted by initiative (highest first)
|
|
current_turn_index: Index in turn_order for current turn
|
|
round_number: Current round (increments each full turn cycle)
|
|
combat_log: History of all actions taken
|
|
status: Current combat status (active, victory, defeat, fled)
|
|
"""
|
|
|
|
encounter_id: str
|
|
combatants: List[Combatant] = field(default_factory=list)
|
|
turn_order: List[str] = field(default_factory=list)
|
|
current_turn_index: int = 0
|
|
round_number: int = 1
|
|
combat_log: List[Dict[str, Any]] = field(default_factory=list)
|
|
status: CombatStatus = CombatStatus.ACTIVE
|
|
|
|
def initialize_combat(self) -> None:
|
|
"""
|
|
Initialize combat by rolling initiative and setting turn order.
|
|
|
|
Initiative: d20 + dexterity bonus
|
|
"""
|
|
# Roll initiative for all combatants
|
|
for combatant in self.combatants:
|
|
# d20 + dexterity bonus
|
|
roll = random.randint(1, 20)
|
|
dex_bonus = combatant.stats.dexterity // 2
|
|
combatant.initiative = roll + dex_bonus
|
|
|
|
# Sort combatants by initiative (highest first)
|
|
sorted_combatants = sorted(self.combatants, key=lambda c: c.initiative, reverse=True)
|
|
self.turn_order = [c.combatant_id for c in sorted_combatants]
|
|
|
|
self.log_action("combat_start", None, f"Combat begins! Round {self.round_number}")
|
|
|
|
def get_current_combatant(self) -> Optional[Combatant]:
|
|
"""Get the combatant whose turn it currently is."""
|
|
if not self.turn_order:
|
|
return None
|
|
|
|
current_id = self.turn_order[self.current_turn_index]
|
|
return self.get_combatant(current_id)
|
|
|
|
def get_combatant(self, combatant_id: str) -> Optional[Combatant]:
|
|
"""Get a combatant by ID."""
|
|
for combatant in self.combatants:
|
|
if combatant.combatant_id == combatant_id:
|
|
return combatant
|
|
return None
|
|
|
|
def advance_turn(self) -> None:
|
|
"""Advance to the next alive combatant's turn, skipping dead combatants."""
|
|
# Track starting position to detect full cycle
|
|
start_index = self.current_turn_index
|
|
rounds_advanced = 0
|
|
|
|
while True:
|
|
self.current_turn_index += 1
|
|
|
|
# If we've cycled through all combatants, start a new round
|
|
if self.current_turn_index >= len(self.turn_order):
|
|
self.current_turn_index = 0
|
|
self.round_number += 1
|
|
rounds_advanced += 1
|
|
self.log_action("round_start", None, f"Round {self.round_number} begins")
|
|
|
|
# Get the current combatant
|
|
current = self.get_current_combatant()
|
|
|
|
# If combatant is alive, their turn starts
|
|
if current and current.is_alive():
|
|
break
|
|
|
|
# Safety check: if we've gone through all combatants twice without finding
|
|
# someone alive, break to avoid infinite loop (combat should end)
|
|
if rounds_advanced >= 2:
|
|
break
|
|
|
|
def start_turn(self) -> List[Dict[str, Any]]:
|
|
"""
|
|
Process the start of a turn.
|
|
|
|
- Tick all effects on current combatant
|
|
- Tick cooldowns
|
|
- Check for stun
|
|
|
|
Returns:
|
|
List of effect tick results
|
|
"""
|
|
combatant = self.get_current_combatant()
|
|
if not combatant:
|
|
return []
|
|
|
|
# Process effects
|
|
effect_results = combatant.tick_effects()
|
|
|
|
# Reduce cooldowns
|
|
combatant.tick_cooldowns()
|
|
|
|
return effect_results
|
|
|
|
def check_end_condition(self) -> CombatStatus:
|
|
"""
|
|
Check if combat should end.
|
|
|
|
Victory: All enemy combatants dead
|
|
Defeat: All player combatants dead
|
|
|
|
Returns:
|
|
Updated combat status
|
|
"""
|
|
players_alive = any(c.is_alive() and c.is_player for c in self.combatants)
|
|
enemies_alive = any(c.is_alive() and not c.is_player for c in self.combatants)
|
|
|
|
if not enemies_alive and players_alive:
|
|
self.status = CombatStatus.VICTORY
|
|
self.log_action("combat_end", None, "Victory! All enemies defeated!")
|
|
elif not players_alive:
|
|
self.status = CombatStatus.DEFEAT
|
|
self.log_action("combat_end", None, "Defeat! All players have fallen!")
|
|
|
|
return self.status
|
|
|
|
def log_action(self, action_type: str, combatant_id: Optional[str], message: str, details: Optional[Dict[str, Any]] = None) -> None:
|
|
"""
|
|
Log a combat action.
|
|
|
|
Args:
|
|
action_type: Type of action (attack, spell, item_use, etc.)
|
|
combatant_id: ID of acting combatant (or None for system messages)
|
|
message: Human-readable message
|
|
details: Additional action details
|
|
"""
|
|
entry = {
|
|
"round": self.round_number,
|
|
"action_type": action_type,
|
|
"combatant_id": combatant_id,
|
|
"message": message,
|
|
"details": details or {},
|
|
}
|
|
self.combat_log.append(entry)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Serialize combat encounter to dictionary."""
|
|
return {
|
|
"encounter_id": self.encounter_id,
|
|
"combatants": [c.to_dict() for c in self.combatants],
|
|
"turn_order": self.turn_order,
|
|
"current_turn_index": self.current_turn_index,
|
|
"round_number": self.round_number,
|
|
"combat_log": self.combat_log,
|
|
"status": self.status.value,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> 'CombatEncounter':
|
|
"""Deserialize combat encounter from dictionary."""
|
|
combatants = [Combatant.from_dict(c) for c in data.get("combatants", [])]
|
|
status = CombatStatus(data.get("status", "active"))
|
|
|
|
return cls(
|
|
encounter_id=data["encounter_id"],
|
|
combatants=combatants,
|
|
turn_order=data.get("turn_order", []),
|
|
current_turn_index=data.get("current_turn_index", 0),
|
|
round_number=data.get("round_number", 1),
|
|
combat_log=data.get("combat_log", []),
|
|
status=status,
|
|
)
|