Combat System Tutorial - Part 3: Creating the System¶
In this part, you will create the CombatSystem that processes attacks, calculates damage, and handles entity death. Systems are where the game logic lives in an ECS architecture.
Overview¶
The CombatSystem will:
- Listen for
AttackRequestEventevents - Validate that combat is possible (same room, valid targets)
- Calculate damage using attack and defense components
- Apply damage to the defender's health
- Emit
DamageDealtEventfor other systems to react - Handle death by emitting
EntityDeathEvent
Step 1: Define Combat Events¶
First, create the events that the combat system will use:
# src/maid_combat_system/events/combat_events.py
"""Combat-related events."""
from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID
from maid_engine.core.events import Event
@dataclass
class AttackRequestEvent(Event):
"""Event requesting an attack between two entities.
This event is typically emitted by the attack command when a player
initiates combat. The CombatSystem listens for this event and processes
the attack.
Attributes:
attacker_id: UUID of the attacking entity
target_id: UUID of the target entity
attack_type: Type of attack (melee, ranged, spell)
"""
attacker_id: UUID
target_id: UUID
attack_type: str = "melee"
@dataclass
class DamageDealtEvent(Event):
"""Event emitted when damage is dealt to an entity.
This event is emitted after damage calculation is complete and the
damage has been applied. Other systems can listen for this to trigger
effects, update UI, or log combat.
Attributes:
source_id: UUID of the entity that dealt damage (None for environmental)
target_id: UUID of the entity that received damage
damage: Amount of damage dealt
damage_type: Type of damage (physical, fire, etc.)
was_critical: Whether this was a critical hit
was_blocked: Whether some damage was blocked by defense
overkill: Amount of damage beyond lethal
"""
source_id: UUID | None
target_id: UUID
damage: int
damage_type: str = "physical"
was_critical: bool = False
was_blocked: bool = False
overkill: int = 0
@dataclass
class CombatMissEvent(Event):
"""Event emitted when an attack misses.
Attributes:
attacker_id: UUID of the attacking entity
target_id: UUID of the target entity
reason: Why the attack missed (missed, evaded)
"""
attacker_id: UUID
target_id: UUID
reason: str = "missed" # "missed" or "evaded"
@dataclass
class EntityDeathEvent(Event):
"""Event emitted when an entity dies.
This event is emitted when an entity's health reaches zero. Other
systems can listen for this to handle death logic, respawning,
loot drops, etc.
Attributes:
entity_id: UUID of the entity that died
killer_id: UUID of the entity that killed it (None for environmental)
death_cause: What caused the death
"""
entity_id: UUID
killer_id: UUID | None = None
death_cause: str = "combat"
Update the events init file:
# src/maid_combat_system/events/__init__.py
"""Combat events.
Events for combat system communication:
- AttackRequestEvent: Request to initiate an attack
- DamageDealtEvent: Notification that damage was dealt
- CombatMissEvent: Notification that an attack missed
- EntityDeathEvent: Notification that an entity died
"""
from .combat_events import (
AttackRequestEvent,
CombatMissEvent,
DamageDealtEvent,
EntityDeathEvent,
)
__all__ = [
"AttackRequestEvent",
"CombatMissEvent",
"DamageDealtEvent",
"EntityDeathEvent",
]
Step 2: Create the Combat System¶
Now create the main combat system:
# src/maid_combat_system/systems/combat_system.py
"""Combat system for processing attacks and damage."""
from __future__ import annotations
import random
from typing import TYPE_CHECKING, ClassVar
from uuid import UUID
from maid_engine.core.ecs import System
from maid_stdlib.components import DescriptionComponent, HealthComponent, PositionComponent
from ..components import AttackComponent, DefenseComponent
from ..events import (
AttackRequestEvent,
CombatMissEvent,
DamageDealtEvent,
EntityDeathEvent,
)
if TYPE_CHECKING:
from maid_engine.core.ecs import Entity
class CombatSystem(System):
"""System that processes combat between entities.
This system handles:
- Attack requests and validation
- Hit/miss determination
- Damage calculation with criticals
- Defense and resistance application
- Death handling
The system uses events for communication:
- Listens for: AttackRequestEvent
- Emits: DamageDealtEvent, CombatMissEvent, EntityDeathEvent
"""
# System priority - combat runs in the middle of the tick
priority: ClassVar[int] = 50
# Combat settings (can be configured)
min_damage: int = 1 # Minimum damage dealt on a hit
async def startup(self) -> None:
"""Initialize the combat system.
Subscribes to attack request events.
"""
self.events.subscribe(AttackRequestEvent, self._handle_attack_request)
async def shutdown(self) -> None:
"""Clean up the combat system."""
# Unsubscribe is handled automatically by the event bus
pass
async def update(self, delta: float) -> None:
"""Process ongoing combat effects.
Currently, all combat is event-driven, so this method doesn't
need to do anything. In the future, this could handle:
- Auto-attack timers
- Damage over time effects
- Combat timeout/disengagement
"""
pass
async def _handle_attack_request(self, event: AttackRequestEvent) -> None:
"""Handle an attack request event.
This method validates the attack, calculates damage, and
applies it to the target.
Args:
event: The attack request event to process
"""
# Get attacker and target entities
attacker = self.entities.get(event.attacker_id)
target = self.entities.get(event.target_id)
# Validate entities exist
if not attacker or not target:
return
# Validate attacker can attack
if not attacker.has(AttackComponent):
return
# Validate target can be damaged
if not target.has(HealthComponent):
return
# Validate entities are in the same room
if not self._in_same_room(attacker, target):
return
# Get target's current health
target_health = target.get(HealthComponent)
# Skip if target is already dead
if not target_health.is_alive:
return
# Process the attack
await self._process_attack(attacker, target, event.attack_type)
async def _process_attack(
self,
attacker: Entity,
target: Entity,
attack_type: str
) -> None:
"""Process an attack from attacker to target.
Args:
attacker: The attacking entity
target: The target entity
attack_type: Type of attack being made
"""
attack = attacker.get(AttackComponent)
defense = target.try_get(DefenseComponent)
target_health = target.get(HealthComponent)
# Step 1: Check for evasion
if defense and self._roll_evasion(defense):
await self.events.emit(CombatMissEvent(
attacker_id=attacker.id,
target_id=target.id,
reason="evaded",
))
return
# Step 2: Check for hit
if not self._roll_hit(attack):
await self.events.emit(CombatMissEvent(
attacker_id=attacker.id,
target_id=target.id,
reason="missed",
))
return
# Step 3: Calculate base damage
base_damage = attack.calculate_base_damage()
# Step 4: Check for critical hit
was_critical = self._roll_critical(attack)
if was_critical:
damage = attack.calculate_critical_damage(base_damage)
else:
damage = base_damage
# Step 5: Apply defenses
if defense:
final_damage = defense.calculate_damage_taken(damage, attack.damage_type)
was_blocked = final_damage < damage
else:
final_damage = damage
was_blocked = False
# Ensure minimum damage
final_damage = max(self.min_damage, final_damage)
# Step 6: Apply damage to target
health_before = target_health.current
actual_damage = target_health.damage(final_damage)
overkill = max(0, final_damage - health_before)
# Step 7: Emit damage event
await self.events.emit(DamageDealtEvent(
source_id=attacker.id,
target_id=target.id,
damage=actual_damage,
damage_type=attack.damage_type,
was_critical=was_critical,
was_blocked=was_blocked,
overkill=overkill,
))
# Step 8: Check for death
if not target_health.is_alive:
await self._handle_death(target, attacker)
async def _handle_death(self, entity: Entity, killer: Entity | None) -> None:
"""Handle an entity's death.
Args:
entity: The entity that died
killer: The entity that killed it (may be None)
"""
await self.events.emit(EntityDeathEvent(
entity_id=entity.id,
killer_id=killer.id if killer else None,
death_cause="combat",
))
def _in_same_room(self, entity1: Entity, entity2: Entity) -> bool:
"""Check if two entities are in the same room.
Args:
entity1: First entity
entity2: Second entity
Returns:
True if both entities are in the same room.
"""
pos1 = entity1.try_get(PositionComponent)
pos2 = entity2.try_get(PositionComponent)
if not pos1 or not pos2:
return False
return pos1.room_id == pos2.room_id
def _roll_hit(self, attack: AttackComponent) -> bool:
"""Roll to see if an attack hits.
Args:
attack: The attacker's attack component
Returns:
True if the attack hits.
"""
roll = random.randint(1, 100)
return attack.roll_hit(roll)
def _roll_critical(self, attack: AttackComponent) -> bool:
"""Roll to see if an attack is a critical hit.
Args:
attack: The attacker's attack component
Returns:
True if the attack is a critical hit.
"""
roll = random.randint(1, 100)
return attack.roll_critical(roll)
def _roll_evasion(self, defense: DefenseComponent) -> bool:
"""Roll to see if the defender evades.
Args:
defense: The defender's defense component
Returns:
True if the attack is evaded.
"""
roll = random.randint(1, 100)
return defense.roll_evasion(roll)
# Public API for direct combat actions
async def request_attack(
self,
attacker_id: UUID,
target_id: UUID,
attack_type: str = "melee"
) -> None:
"""Request an attack between two entities.
This is a convenience method that emits an AttackRequestEvent.
Args:
attacker_id: UUID of the attacker
target_id: UUID of the target
attack_type: Type of attack
"""
await self.events.emit(AttackRequestEvent(
attacker_id=attacker_id,
target_id=target_id,
attack_type=attack_type,
))
def get_damage_preview(
self,
attacker_id: UUID,
target_id: UUID
) -> dict[str, int | str] | None:
"""Get a preview of potential damage without attacking.
Useful for UI to show damage ranges.
Args:
attacker_id: UUID of the potential attacker
target_id: UUID of the potential target
Returns:
Dictionary with damage preview info, or None if attack not possible.
"""
attacker = self.entities.get(attacker_id)
target = self.entities.get(target_id)
if not attacker or not target:
return None
attack = attacker.try_get(AttackComponent)
if not attack:
return None
defense = target.try_get(DefenseComponent)
base_damage = attack.calculate_base_damage()
crit_damage = attack.calculate_critical_damage(base_damage)
# Calculate expected damage after defense
if defense:
min_after_defense = defense.calculate_damage_taken(
base_damage, attack.damage_type
)
max_after_defense = defense.calculate_damage_taken(
crit_damage, attack.damage_type
)
else:
min_after_defense = base_damage
max_after_defense = crit_damage
return {
"min_damage": min_after_defense,
"max_damage": max_after_defense,
"hit_chance": attack.accuracy,
"crit_chance": attack.critical_chance,
"damage_type": attack.damage_type,
}
Update the systems init file:
# src/maid_combat_system/systems/__init__.py
"""Combat systems.
Systems for processing combat:
- CombatSystem: Main combat processing (attacks, damage, death)
"""
from .combat_system import CombatSystem
__all__ = ["CombatSystem"]
Step 3: Understanding the Combat Flow¶
Here's the complete flow when an attack is requested:
Player types "attack goblin"
|
v
+-------------------+
| Attack Command | <-- Validates target, emits event
+-------------------+
|
| AttackRequestEvent
v
+-------------------+
| CombatSystem | <-- Listens for event
| _handle_attack() |
+-------------------+
|
| Validate: entities exist
| Validate: same room
| Validate: target alive
v
+-------------------+
| _process_attack() |
+-------------------+
|
+--> Roll evasion --> [evaded] --> CombatMissEvent
|
+--> Roll hit --> [missed] --> CombatMissEvent
|
| [hit]
v
+-------------------+
| Calculate damage |
| - Base damage |
| - Critical? |
| - Apply defense |
+-------------------+
|
v
+-------------------+
| Apply damage |
| target.damage() |
+-------------------+
|
| DamageDealtEvent
v
+-------------------+
| Check death |
+-------------------+
|
| [dead] --> EntityDeathEvent
v
[done]
Step 4: Testing the System¶
Create comprehensive tests for the combat system:
# tests/test_system.py
"""Tests for the combat system."""
import pytest
from unittest.mock import patch
from uuid import uuid4
from maid_stdlib.components import HealthComponent, PositionComponent
from maid_combat_system.components import AttackComponent, DefenseComponent
from maid_combat_system.events import (
AttackRequestEvent,
CombatMissEvent,
DamageDealtEvent,
EntityDeathEvent,
)
from maid_combat_system.systems import CombatSystem
@pytest.fixture
def combat_system(world):
"""Create a combat system for testing."""
system = CombatSystem(world)
return system
@pytest.fixture
async def started_combat_system(combat_system):
"""Create and start a combat system."""
await combat_system.startup()
return combat_system
@pytest.fixture
def attacker_entity(world, room_id):
"""Create an attacker entity with combat components."""
entity = world.entities.create()
entity.add(PositionComponent(room_id=room_id))
entity.add(HealthComponent(current=100, maximum=100))
entity.add(AttackComponent(
attack_power=20,
accuracy=100, # Always hits for testing
critical_chance=0, # No crits for predictable testing
))
return entity
@pytest.fixture
def defender_entity(world, room_id):
"""Create a defender entity with combat components."""
entity = world.entities.create()
entity.add(PositionComponent(room_id=room_id))
entity.add(HealthComponent(current=50, maximum=50))
entity.add(DefenseComponent(
defense=5,
evasion=0, # No evasion for predictable testing
))
return entity
class TestCombatSystem:
"""Tests for CombatSystem."""
@pytest.mark.asyncio
async def test_system_startup(self, combat_system, world):
"""Test system subscribes to events on startup."""
await combat_system.startup()
# System should be subscribed to AttackRequestEvent
assert world.events.has_handlers(AttackRequestEvent)
@pytest.mark.asyncio
async def test_basic_attack_deals_damage(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that an attack deals damage to the target."""
defender_health = defender_entity.get(HealthComponent)
initial_health = defender_health.current
# Emit attack request
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
# Health should be reduced
# Attack: 20, Defense: 5, Expected damage: 15
assert defender_health.current == initial_health - 15
@pytest.mark.asyncio
async def test_attack_emits_damage_event(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that attack emits DamageDealtEvent."""
events_received = []
async def capture_event(event: DamageDealtEvent):
events_received.append(event)
world.events.subscribe(DamageDealtEvent, capture_event)
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
assert len(events_received) == 1
assert events_received[0].source_id == attacker_entity.id
assert events_received[0].target_id == defender_entity.id
assert events_received[0].damage == 15
@pytest.mark.asyncio
async def test_miss_emits_miss_event(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that a miss emits CombatMissEvent."""
# Set accuracy to 0 so attack always misses
attack = attacker_entity.get(AttackComponent)
attack.accuracy = 0
events_received = []
async def capture_event(event: CombatMissEvent):
events_received.append(event)
world.events.subscribe(CombatMissEvent, capture_event)
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
assert len(events_received) == 1
assert events_received[0].reason == "missed"
@pytest.mark.asyncio
async def test_evasion_emits_miss_event(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that evasion emits CombatMissEvent."""
# Set evasion to 100 so target always evades
defense = defender_entity.get(DefenseComponent)
defense.evasion = 100
events_received = []
async def capture_event(event: CombatMissEvent):
events_received.append(event)
world.events.subscribe(CombatMissEvent, capture_event)
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
assert len(events_received) == 1
assert events_received[0].reason == "evaded"
@pytest.mark.asyncio
async def test_death_emits_death_event(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that killing a target emits EntityDeathEvent."""
# Set defender health low enough to die
health = defender_entity.get(HealthComponent)
health.current = 10 # Will die from 15 damage
events_received = []
async def capture_event(event: EntityDeathEvent):
events_received.append(event)
world.events.subscribe(EntityDeathEvent, capture_event)
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
assert len(events_received) == 1
assert events_received[0].entity_id == defender_entity.id
assert events_received[0].killer_id == attacker_entity.id
@pytest.mark.asyncio
async def test_cannot_attack_different_room(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that attacks fail across rooms."""
# Move defender to different room
defender_pos = defender_entity.get(PositionComponent)
defender_pos.room_id = uuid4()
defender_health = defender_entity.get(HealthComponent)
initial_health = defender_health.current
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
# Health should not change
assert defender_health.current == initial_health
@pytest.mark.asyncio
async def test_cannot_attack_dead_target(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that attacks fail on dead targets."""
# Kill the defender
health = defender_entity.get(HealthComponent)
health.current = 0
events_received = []
async def capture_event(event: DamageDealtEvent):
events_received.append(event)
world.events.subscribe(DamageDealtEvent, capture_event)
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
# No damage event should be emitted
assert len(events_received) == 0
@pytest.mark.asyncio
async def test_critical_hit_multiplies_damage(
self,
started_combat_system,
world,
attacker_entity,
defender_entity
):
"""Test that critical hits multiply damage."""
# Remove defender's defense for easier calculation
defender_entity.remove(DefenseComponent)
# Set critical chance to 100%
attack = attacker_entity.get(AttackComponent)
attack.critical_chance = 100
attack.critical_multiplier = 2.0
events_received = []
async def capture_event(event: DamageDealtEvent):
events_received.append(event)
world.events.subscribe(DamageDealtEvent, capture_event)
await world.events.emit(AttackRequestEvent(
attacker_id=attacker_entity.id,
target_id=defender_entity.id,
))
assert len(events_received) == 1
assert events_received[0].was_critical is True
# Base damage 20 * 2.0 crit multiplier = 40
assert events_received[0].damage == 40
@pytest.mark.asyncio
async def test_damage_preview(
self,
combat_system,
world,
attacker_entity,
defender_entity
):
"""Test damage preview calculation."""
preview = combat_system.get_damage_preview(
attacker_entity.id,
defender_entity.id
)
assert preview is not None
assert preview["min_damage"] == 15 # 20 - 5 defense
assert preview["hit_chance"] == 100
assert preview["crit_chance"] == 0
Run the tests:
Step 5: Extending the System¶
Here are some ways to extend the combat system:
Auto-Attack Support¶
Add auto-attack in the update method:
async def update(self, delta: float) -> None:
"""Process auto-attacks for entities in combat."""
for entity in self.entities.with_components(AttackComponent):
attack = entity.get(AttackComponent)
# Check if in combat and has a target
if not attack.in_combat or not attack.target_id:
continue
# Check attack cooldown
attack.time_since_attack += delta
if attack.time_since_attack >= (1.0 / attack.attack_speed):
attack.time_since_attack = 0.0
await self.request_attack(entity.id, attack.target_id)
Damage Over Time¶
Create a poison/bleed system:
@dataclass
class DamageOverTimeComponent(Component):
"""Component for ongoing damage effects."""
damage_per_second: int
remaining_duration: float
damage_type: str = "poison"
source_id: UUID | None = None
async def update(self, delta: float) -> None:
"""Process damage over time effects."""
for entity in self.entities.with_components(
DamageOverTimeComponent, HealthComponent
):
dot = entity.get(DamageOverTimeComponent)
health = entity.get(HealthComponent)
# Calculate damage for this tick
damage = int(dot.damage_per_second * delta)
if damage > 0:
health.damage(damage)
await self.events.emit(DamageDealtEvent(
source_id=dot.source_id,
target_id=entity.id,
damage=damage,
damage_type=dot.damage_type,
))
# Reduce duration
dot.remaining_duration -= delta
if dot.remaining_duration <= 0:
entity.remove(DamageOverTimeComponent)
Summary¶
In this part, you:
- Created combat events for system communication
- Built the
CombatSystemwith attack processing - Implemented damage calculation with criticals and defense
- Added death handling with events
- Created comprehensive tests for the system
Next Steps¶
In Part 4: Commands, you will create the player-facing commands:
attackcommand for initiating combat- Target selection and validation
- Combat feedback messages
Continue to Part 4: Commands