Systems Guide

Systems contain the game logic that operates on entities. They are the "S" in ECS (Entity Component System) and are where you implement behaviors like combat, movement, AI, and more.

System Basics

Creating a System

Systems inherit from the System base class and implement the update method:

from maid_engine.core.ecs import System


class RegenerationSystem(System):
    """Regenerates health for living entities."""

    async def update(self, delta: float) -> None:
        """Called every game tick.

        Args:
            delta: Time since last update in seconds
        """
        for entity in self.entities.with_components(HealthComponent):
            health = entity.get(HealthComponent)
            if health.is_alive and health.current < health.maximum:
                regen = health.regeneration_rate * delta
                health.heal(int(regen))

System Properties

Systems have access to important properties:

class MySystem(System):
    async def update(self, delta: float) -> None:
        # Access the world
        self.world  # The game World instance

        # Access the entity manager
        self.entities  # EntityManager for queries

        # Access the event bus
        self.events  # EventBus for pub/sub

System Priority

Systems run in priority order each tick. Lower priority numbers run earlier:

class MovementSystem(System):
    priority = 10  # Runs early


class CombatSystem(System):
    priority = 50  # Runs in the middle


class CleanupSystem(System):
    priority = 200  # Runs late

Common priority ranges:

Range	Purpose
0-20	Input processing, command handling
20-50	Core game logic (movement, combat)
50-100	Secondary systems (status effects, AI)
100-150	Output, notifications
150+	Cleanup, state persistence

Entity Queries

Systems typically query for entities with specific component combinations:

Single Component Query

for entity in self.entities.with_components(HealthComponent):
    health = entity.get(HealthComponent)
    # Process entity...

Multiple Component Query

for entity in self.entities.with_components(
    PositionComponent,
    HealthComponent,
    CombatComponent
):
    pos = entity.get(PositionComponent)
    health = entity.get(HealthComponent)
    combat = entity.get(CombatComponent)
    # Process entity...

Tag-Based Query

# Find all entities with a specific tag
for entity in self.entities.with_tag("hostile"):
    # Process hostile entities...

# Find entities with multiple tags
for entity in self.entities.with_tags("player", "online"):
    # Process online players...

Optional Components

for entity in self.entities.with_components(HealthComponent):
    health = entity.get(HealthComponent)

    # Check for optional components
    if entity.has(CombatComponent):
        combat = entity.get(CombatComponent)
        # Apply combat modifiers...

    # Or use try_get for optional components
    shield = entity.try_get(ShieldComponent)
    if shield:
        # Apply shield effects...

Lifecycle Hooks

Systems have startup and shutdown hooks for initialization and cleanup:

class CombatSystem(System):
    async def startup(self) -> None:
        """Called when the system starts."""
        # Subscribe to events
        self.events.subscribe(CombatStartEvent, self._on_combat_start)
        self.events.subscribe(DamageDealtEvent, self._on_damage_dealt)

        # Initialize state
        self._active_combats: dict[UUID, CombatState] = {}

    async def shutdown(self) -> None:
        """Called when the system shuts down."""
        # Clean up active combats
        for combat_state in self._active_combats.values():
            await self._end_combat(combat_state)
        self._active_combats.clear()

    async def update(self, delta: float) -> None:
        # Process combat...
        pass

    async def _on_combat_start(self, event: CombatStartEvent) -> None:
        # Handle combat initiation...
        pass

    async def _on_damage_dealt(self, event: DamageDealtEvent) -> None:
        # Handle damage processing...
        pass

Enabling and Disabling Systems

Systems can be enabled or disabled at runtime:

class WeatherSystem(System):
    enabled: bool = True  # Default enabled

    async def update(self, delta: float) -> None:
        if not self.enabled:
            return  # Base class checks this, but explicit check is fine too

        # Update weather...

Control from outside:

# Disable a system
world.systems.disable(WeatherSystem)

# Enable a system
world.systems.enable(WeatherSystem)

# Check if enabled
system = world.systems.get(WeatherSystem)
if system and system.enabled:
    # System is active

Emitting Events

Systems communicate through events rather than direct calls:

class CombatSystem(System):
    async def update(self, delta: float) -> None:
        for entity in self.entities.with_components(CombatComponent):
            combat = entity.get(CombatComponent)

            if combat.in_combat and combat.target_id:
                # Calculate damage
                damage = self._calculate_damage(entity, combat.target_id)

                # Emit damage event
                await self.events.emit(DamageDealtEvent(
                    source_id=entity.id,
                    target_id=combat.target_id,
                    damage=damage,
                    damage_type="physical",
                ))

Time-Based Updates

Use the delta parameter for time-based calculations:

class PoisonSystem(System):
    async def update(self, delta: float) -> None:
        for entity in self.entities.with_components(PoisonComponent, HealthComponent):
            poison = entity.get(PoisonComponent)
            health = entity.get(HealthComponent)

            # Accumulate time
            poison.time_since_tick += delta

            # Tick poison every second
            if poison.time_since_tick >= 1.0:
                poison.time_since_tick -= 1.0
                poison.remaining_duration -= 1.0

                # Apply poison damage
                damage = poison.damage_per_second
                health.damage(damage)

                await self.events.emit(DamageDealtEvent(
                    source_id=None,
                    target_id=entity.id,
                    damage=damage,
                    damage_type="poison",
                ))

                # Remove expired poison
                if poison.remaining_duration <= 0:
                    entity.remove(PoisonComponent)

System Dependencies

If your system depends on another system, declare it through priority ordering:

class MovementSystem(System):
    priority = 20  # Runs first


class CollisionSystem(System):
    priority = 25  # Runs after movement

    async def update(self, delta: float) -> None:
        # Movement has already been processed
        # Now check for collisions...

Or query the other system directly:

class AISystem(System):
    priority = 60

    async def update(self, delta: float) -> None:
        # Get reference to combat system
        combat_system = self.world.systems.get(CombatSystem)
        if combat_system:
            # Use combat system data...
            pass

Example: Complete Combat System

Here's a more complete example of a combat system:

from dataclasses import dataclass
from typing import ClassVar
from uuid import UUID

from maid_engine.core.ecs import System
from maid_engine.core.events import Event
from maid_stdlib.components import CombatComponent, HealthComponent, PositionComponent


@dataclass
class AttackEvent(Event):
    """Request to attack a target."""
    attacker_id: UUID
    target_id: UUID
    attack_type: str = "melee"


@dataclass
class DamageDealtEvent(Event):
    """Damage was dealt to a target."""
    source_id: UUID | None
    target_id: UUID
    damage: int
    damage_type: str
    was_critical: bool = False


class CombatSystem(System):
    """Handles combat between entities."""

    priority: ClassVar[int] = 50

    async def startup(self) -> None:
        """Subscribe to combat-related events."""
        self.events.subscribe(AttackEvent, self._handle_attack)

    async def _handle_attack(self, event: AttackEvent) -> None:
        """Process an attack request."""
        attacker = self.entities.get(event.attacker_id)
        target = self.entities.get(event.target_id)

        if not attacker or not target:
            return

        # Verify both have required components
        if not attacker.has(CombatComponent, PositionComponent):
            return
        if not target.has(HealthComponent, PositionComponent):
            return

        attacker_pos = attacker.get(PositionComponent)
        target_pos = target.get(PositionComponent)

        # Check same room
        if attacker_pos.room_id != target_pos.room_id:
            return

        attacker_combat = attacker.get(CombatComponent)
        target_health = target.get(HealthComponent)

        # Calculate hit chance
        hit_roll = random.randint(1, 100)
        if hit_roll > attacker_combat.accuracy:
            # Miss
            return

        # Check for dodge
        dodge_roll = random.randint(1, 100)
        target_combat = target.try_get(CombatComponent)
        evasion = target_combat.evasion if target_combat else 0
        if dodge_roll <= evasion:
            # Dodged
            return

        # Calculate damage
        base_damage = attacker_combat.attack_power
        defense = target_combat.defense if target_combat else 0
        damage = max(1, base_damage - defense)

        # Check for critical hit
        was_critical = random.randint(1, 100) <= attacker_combat.critical_chance
        if was_critical:
            damage = int(damage * attacker_combat.critical_multiplier)

        # Apply damage
        actual_damage = target_health.damage(damage)

        # Emit damage event
        await self.events.emit(DamageDealtEvent(
            source_id=attacker.id,
            target_id=target.id,
            damage=actual_damage,
            damage_type="physical",
            was_critical=was_critical,
        ))

        # Check for death
        if not target_health.is_alive:
            await self.events.emit(EntityDeathEvent(
                entity_id=target.id,
                killer_id=attacker.id,
            ))

    async def update(self, delta: float) -> None:
        """Process ongoing combat each tick."""
        for entity in self.entities.with_components(CombatComponent):
            combat = entity.get(CombatComponent)

            if combat.in_combat and combat.target_id:
                # Auto-attack logic could go here
                pass

Best Practices

Keep Systems Focused

Each system should have a single responsibility:

# Good: Separate focused systems
class HealthRegenerationSystem(System): ...
class ManaRegenerationSystem(System): ...
class StaminaRegenerationSystem(System): ...

# Avoid: One system doing too much
class AllRegenerationSystem(System): ...  # Too broad

Use Events for Communication

Prefer events over direct system coupling:

# Good: Emit events
await self.events.emit(DamageDealtEvent(...))

# Avoid: Direct system calls
other_system = self.world.systems.get(OtherSystem)
other_system.some_method()  # Creates tight coupling

Handle Missing Entities

Always check that entities and components exist:

async def update(self, delta: float) -> None:
    for entity in self.entities.with_components(MyComponent):
        # Safe: entity guaranteed to have MyComponent

        # Still check optional components
        optional = entity.try_get(OptionalComponent)
        if optional:
            # Use optional component

Minimize Work Per Tick

Systems run every tick, so keep update() efficient:

class OptimizedSystem(System):
    async def update(self, delta: float) -> None:
        # Cache expensive lookups
        if not self._cache_valid:
            self._rebuild_cache()

        # Process only what's needed
        for entity in self._cached_entities:
            # Quick processing...

Next Steps

• Commands Guide - Creating player commands
• Events Guide - Working with the event bus
• ECS Overview - Deep dive into ECS architecture