Systems in MAID¶
Systems contain the logic that operates on entities with specific components. They are executed each game tick by the SystemManager.
What is a System?¶
A system is a class that:
- Queries entities with specific components
- Processes those entities each tick
- May emit events or modify component data
from maid_engine.core.ecs import System
class HealthRegenSystem(System):
priority = 100
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(HealthComponent):
health = entity.get(HealthComponent)
if health.current < health.maximum:
health.current = min(
health.current + delta * 0.5,
health.maximum
)
Creating Systems¶
Basic System¶
from maid_engine.core.ecs import System
from maid_engine.core.world import World
class MovementSystem(System):
"""Processes entity movement each tick."""
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(
PositionComponent, VelocityComponent
):
pos = entity.get(PositionComponent)
vel = entity.get(VelocityComponent)
pos.x += vel.dx * delta
pos.y += vel.dy * delta
System Priority¶
Priority determines execution order (lower values run first):
class InputSystem(System):
priority = 10 # Runs early - process input first
class PhysicsSystem(System):
priority = 50 # Runs in the middle
class RenderSystem(System):
priority = 100 # Runs late - after all updates
Enabling/Disabling Systems¶
class OptionalSystem(System):
enabled = True # Can be toggled
async def update(self, delta: float) -> None:
# Only runs when enabled
pass
# Toggle from outside
system = world.systems.get(OptionalSystem)
system.enabled = False
System Properties¶
Systems have access to the world and its subsystems:
class MySystem(System):
async def update(self, delta: float) -> None:
# Access the world
world = self.world
# Access the entity manager
entities = self.entities
# Access the event bus
events = self.events
# Query entities
for entity in self.entities.with_components(SomeComponent):
pass
Lifecycle Methods¶
startup()¶
Called when the system starts:
class DatabaseSystem(System):
async def startup(self) -> None:
"""Initialize database connection."""
self._connection = await create_connection()
self._cache = {}
async def update(self, delta: float) -> None:
# Use self._connection
pass
shutdown()¶
Called when the system stops:
class DatabaseSystem(System):
async def shutdown(self) -> None:
"""Clean up database connection."""
if self._connection:
await self._connection.close()
update()¶
Called every game tick:
class TickingSystem(System):
async def update(self, delta: float) -> None:
"""Process one tick.
Args:
delta: Time since last tick in seconds
"""
# delta is typically 0.25 seconds (4 ticks per second)
self._elapsed += delta
for entity in self.entities.with_components(TimerComponent):
timer = entity.get(TimerComponent)
timer.remaining -= delta
Registering Systems¶
Via Content Pack¶
The recommended way:
class MyContentPack:
def get_systems(self, world: World) -> list[System]:
return [
MovementSystem(world),
CombatSystem(world),
HealthRegenSystem(world),
]
Via SystemManager¶
Direct registration:
With Dependencies¶
Specify system dependencies:
world.systems.register(
CombatSystem(world),
source_pack="my-pack",
dependencies=[MovementSystem, TargetingSystem],
)
Querying Entities¶
By Components¶
class CombatSystem(System):
async def update(self, delta: float) -> None:
# Entities with all specified components
for entity in self.entities.with_components(
HealthComponent, CombatStatsComponent
):
health = entity.get(HealthComponent)
stats = entity.get(CombatStatsComponent)
# Process combat
By Tags¶
class PlayerSystem(System):
async def update(self, delta: float) -> None:
for entity in self.entities.with_tag("player"):
# Process player-specific logic
pass
for entity in self.entities.with_tags("npc", "hostile"):
# Process hostile NPCs
pass
Combined Queries¶
class AISystem(System):
async def update(self, delta: float) -> None:
# Get entities with AI component
for entity in self.entities.with_components(AIComponent):
# Skip players (they're controlled by users)
if entity.has_tag("player"):
continue
ai = entity.get(AIComponent)
await self._process_ai(entity, ai)
Emitting Events¶
Systems communicate via events:
from maid_engine.core.events import Event
from dataclasses import dataclass
@dataclass
class EntityDamagedEvent(Event):
entity_id: UUID
damage: int
source_id: UUID | None = None
class CombatSystem(System):
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(InCombatComponent):
damage = self._calculate_damage(entity)
if damage > 0:
await self.events.emit(EntityDamagedEvent(
entity_id=entity.id,
damage=damage,
source_id=attacker.id,
))
Subscribing to Events¶
class DeathSystem(System):
async def startup(self) -> None:
# Subscribe to damage events
self._handler_id = self.events.subscribe(
EntityDamagedEvent,
self._on_damage,
)
async def shutdown(self) -> None:
# Clean up subscription
self.events.unsubscribe(self._handler_id)
async def _on_damage(self, event: EntityDamagedEvent) -> None:
entity = self.entities.get(event.entity_id)
if entity:
health = entity.try_get(HealthComponent)
if health and health.current <= 0:
await self._handle_death(entity)
async def update(self, delta: float) -> None:
# May not need to do anything in update
pass
StatefulSystem Protocol¶
For systems that need to preserve state across hot reloads:
from typing import Any
from maid_engine.plugins.migration import StatefulSystem
class QueueSystem(System):
"""System that maintains a queue across reloads."""
def __init__(self, world: World) -> None:
super().__init__(world)
self._queue: list[dict] = []
self._processed: int = 0
def capture_state(self) -> dict[str, Any]:
"""Capture state before hot reload."""
return {
"queue": self._queue.copy(),
"processed": self._processed,
}
def restore_state(self, state: dict[str, Any]) -> None:
"""Restore state after hot reload."""
self._queue = state.get("queue", [])
self._processed = state.get("processed", 0)
async def update(self, delta: float) -> None:
while self._queue:
item = self._queue.pop(0)
await self._process(item)
self._processed += 1
System Management¶
Get Registered Systems¶
# Get a specific system
combat_system = world.systems.get(CombatSystem)
# Get or raise
combat_system = world.systems.get_or_raise(CombatSystem)
# List all systems
for system in world.systems.systems:
print(f"{type(system).__name__}: priority={system.priority}")
Enable/Disable Systems¶
# Disable a system
world.systems.disable(CombatSystem)
# Enable a system
world.systems.enable(CombatSystem)
Unregister Systems¶
# Unregister a single system
world.systems.unregister(CombatSystem)
# Unregister all systems from a pack
removed = await world.systems.unregister_systems_for_pack("my-pack")
Safe System Removal¶
Use the removal context for thread-safe removal:
async with world.systems.safe_removal_context() as ctx:
await ctx.remove_system(CombatSystem)
await ctx.remove_system(AISystem)
# Or for a single system
removed = await world.systems.remove_system_safely(CombatSystem)
Best Practices¶
1. Keep Systems Focused¶
# Good - focused system
class HealthRegenSystem(System):
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(HealthComponent):
# Only health regeneration logic
pass
# Bad - too many responsibilities
class CharacterSystem(System):
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(CharacterComponent):
self._handle_health(entity)
self._handle_mana(entity)
self._handle_movement(entity)
self._handle_combat(entity)
# Too much!
2. Use Priority Wisely¶
# Input processing should happen first
class InputSystem(System):
priority = 10
# Game logic in the middle
class GameLogicSystem(System):
priority = 50
# Cleanup/finalization last
class CleanupSystem(System):
priority = 200
3. Minimize Component Queries¶
# Good - query once, iterate once
class EfficientsSystem(System):
async def update(self, delta: float) -> None:
entities = list(self.entities.with_components(
HealthComponent, StatusEffectsComponent
))
for entity in entities:
# Process all relevant entities together
pass
# Bad - redundant queries
class InefficientSystem(System):
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(HealthComponent):
# Query again?
if entity in self.entities.with_components(StatusEffectsComponent):
pass
4. Handle Missing Components Gracefully¶
class SafeSystem(System):
async def update(self, delta: float) -> None:
for entity in self.entities.with_components(HealthComponent):
health = entity.get(HealthComponent)
# Optional component - use try_get
regen = entity.try_get(RegenerationComponent)
if regen:
health.current += regen.rate * delta
5. Clean Up Resources¶
class ResourcefulSystem(System):
async def startup(self) -> None:
self._resources = await acquire_resources()
self._event_handler = self.events.subscribe(
SomeEvent, self._handle_event
)
async def shutdown(self) -> None:
# Always clean up!
await release_resources(self._resources)
self.events.unsubscribe(self._event_handler)
6. Document System Behavior¶
class CombatSystem(System):
"""Processes combat between entities.
This system handles:
- Damage calculation based on stats
- Attack timing and cooldowns
- Death detection and entity cleanup
Requires components:
- HealthComponent: Target must have health
- CombatStatsComponent: Attacker needs combat stats
- TargetComponent: Attacker needs a target
Emits events:
- EntityDamagedEvent: When damage is dealt
- EntityDiedEvent: When an entity's health reaches 0
"""
priority = 60
Example: Complete Combat System¶
from uuid import UUID
from dataclasses import dataclass
from maid_engine.core.ecs import System
from maid_engine.core.events import Event
@dataclass
class AttackEvent(Event):
attacker_id: UUID
target_id: UUID
damage: int
@dataclass
class EntityDiedEvent(Event):
entity_id: UUID
killer_id: UUID | None = None
class CombatSystem(System):
"""Handles combat between entities."""
priority = 60
def __init__(self, world) -> None:
super().__init__(world)
self._attack_cooldowns: dict[UUID, float] = {}
async def startup(self) -> None:
self._attack_cooldowns = {}
async def shutdown(self) -> None:
self._attack_cooldowns.clear()
async def update(self, delta: float) -> None:
# Update cooldowns
for entity_id in list(self._attack_cooldowns.keys()):
self._attack_cooldowns[entity_id] -= delta
if self._attack_cooldowns[entity_id] <= 0:
del self._attack_cooldowns[entity_id]
# Process attackers
for entity in self.entities.with_components(
CombatStatsComponent, TargetComponent
):
target_comp = entity.get(TargetComponent)
target = self.entities.get(target_comp.target_id)
if not target or entity.id in self._attack_cooldowns:
continue
await self._perform_attack(entity, target)
async def _perform_attack(self, attacker, target) -> None:
stats = attacker.get(CombatStatsComponent)
target_health = target.try_get(HealthComponent)
if not target_health:
return
# Calculate damage
damage = max(1, stats.attack - (
target.try_get(CombatStatsComponent) or 0
))
# Apply damage
target_health.current -= damage
# Set cooldown
self._attack_cooldowns[attacker.id] = 1.0 / stats.speed
# Emit event
await self.events.emit(AttackEvent(
attacker_id=attacker.id,
target_id=target.id,
damage=damage,
))
# Check for death
if target_health.current <= 0:
await self.events.emit(EntityDiedEvent(
entity_id=target.id,
killer_id=attacker.id,
))
# StatefulSystem implementation for hot reload
def capture_state(self) -> dict:
return {
"cooldowns": {
str(k): v for k, v in self._attack_cooldowns.items()
}
}
def restore_state(self, state: dict) -> None:
self._attack_cooldowns = {
UUID(k): v for k, v in state.get("cooldowns", {}).items()
}
Next Steps¶
- Entities - Learn about creating and managing entities
- Components - Learn about designing components
- Events Overview - Learn about the event system