Timed Event¶

Problem¶

You want a recurring world event that fires on a schedule — for example, a volcano that erupts every 10 minutes, affecting all players in the area.

Solution¶

The Event¶

from dataclasses import dataclass
from uuid import UUID
from maid_engine.core.events import Event


@dataclass
class VolcanoEruptionEvent(Event):
    """Fired when the volcano erupts."""
    volcano_room_id: UUID
    affected_room_ids: list[UUID]
    intensity: float = 1.0  # 0.0 to 1.0

The Timer System¶

import random
from uuid import UUID

from maid_engine.core.ecs import System
from maid_engine.core.world import World
from maid_stdlib.components import HealthComponent, DescriptionComponent
from maid_stdlib.events import MessageEvent


class VolcanoSystem(System):
    """Periodically erupts a volcano, damaging nearby entities."""

    priority = 200  # Run late — after combat/movement

    def __init__(
        self,
        world: World,
        volcano_room_id: UUID,
        blast_room_ids: list[UUID],
        interval: float = 600.0,
        damage: int = 15,
    ) -> None:
        super().__init__(world)
        self.volcano_room_id = volcano_room_id
        self.blast_room_ids = blast_room_ids
        self.interval = interval  # Seconds between eruptions
        self.damage = damage
        self._timer: float = interval
        self._warning_sent: bool = False
        self._warning_time: float = 30.0  # Warn 30s before

    async def update(self, delta: float) -> None:
        self._timer -= delta

        # Send warning before eruption
        if (
            not self._warning_sent
            and self._timer <= self._warning_time
        ):
            self._warning_sent = True
            await self._warn_players()

        # Erupt!
        if self._timer <= 0:
            await self._erupt()
            self._timer = self.interval
            self._warning_sent = False

    async def _warn_players(self) -> None:
        """Send warning to all players in affected rooms."""
        target_ids: list[UUID] = []
        all_rooms = [self.volcano_room_id] + self.blast_room_ids
        for room_id in all_rooms:
            for entity in self.world.entities_in_room(room_id):
                if entity.has_tag("player"):
                    target_ids.append(entity.id)
        if target_ids:
            await self.events.emit(MessageEvent(
                sender_id=None,
                target_ids=target_ids,
                channel="room",
                message="The ground rumbles ominously...",
            ))

    async def _erupt(self) -> None:
        """Damage all entities in the blast zone."""
        all_rooms = [self.volcano_room_id] + self.blast_room_ids

        await self.events.emit(VolcanoEruptionEvent(
            volcano_room_id=self.volcano_room_id,
            affected_room_ids=all_rooms,
        ))

        for room_id in all_rooms:
            for entity in self.world.entities_in_room(room_id):
                health = entity.try_get(HealthComponent)
                if not health:
                    continue

                # More damage closer to volcano
                if room_id == self.volcano_room_id:
                    actual_damage = self.damage * 2
                else:
                    actual_damage = self.damage + random.randint(0, 5)

                health.damage(actual_damage)

A Generic Recurring Timer¶

For simpler cases, here's a reusable timer pattern:

from collections.abc import Awaitable, Callable
from maid_engine.core.ecs import System
from maid_engine.core.world import World


class RecurringTimerSystem(System):
    """Generic system that calls a callback on a fixed interval."""

    priority = 250

    def __init__(
        self,
        world: World,
        interval: float,
        callback: Callable[[], Awaitable[None]],
        name: str = "timer",
    ) -> None:
        super().__init__(world)
        self.interval = interval
        self.callback = callback
        self.name = name
        self._elapsed: float = 0.0

    async def update(self, delta: float) -> None:
        self._elapsed += delta
        if self._elapsed >= self.interval:
            self._elapsed -= self.interval
            await self.callback()

Usage:

from functools import partial

from maid_engine.core.world import World
from maid_stdlib.components import HealthComponent

async def blood_moon_event(world: World) -> None:
    """Buff all hostile NPCs during the blood moon."""
    for entity in world.entities.with_tag("hostile"):
        health = entity.try_get(HealthComponent)
        if health:
            health.maximum = int(health.maximum * 1.5)
            health.heal(health.maximum - health.current)


# In get_systems():
systems = [
    RecurringTimerSystem(
        world,
        interval=1800.0,  # Every 30 minutes
        callback=partial(blood_moon_event, world),
        name="blood_moon",
    ),
]

Wiring into a Content Pack¶

from maid_engine.core.ecs import System
from maid_engine.core.world import World
from maid_engine.plugins import ContentPackManifest


class MyContentPack:
    @property
    def manifest(self) -> ContentPackManifest:
        return ContentPackManifest(
            name="my-pack",
            version="1.0.0",
            description="My custom content pack",
        )

    def get_systems(self, world: World) -> list[System]:
        volcano_room = ...  # Your volcano room UUID
        blast_zone = [...]  # Nearby room UUIDs
        return [
            VolcanoSystem(
                world,
                volcano_room_id=volcano_room,
                blast_room_ids=blast_zone,
                interval=600.0,
                damage=15,
            ),
        ]

    def get_events(self) -> list[type]:
        return [VolcanoEruptionEvent]

How It Works¶

System.update(delta) is called every tick with the elapsed time in seconds
The system accumulates delta in a timer and fires when the threshold is reached
Emitting VolcanoEruptionEvent lets other systems react (e.g., update room descriptions, play sounds)
world.entities_in_room() efficiently queries all entities in the blast radius
HealthComponent.damage() handles clamping HP to zero

Variations¶

Random timing: Add random.uniform(-60, 60) to the interval each cycle
Escalating danger: Increase intensity with each eruption over time
Player-triggered: Instead of a timer, erupt when a player pulls a lever (command-triggered)
Day/night cycle: Use RecurringTimerSystem with a 24-minute interval (1 minute = 1 game hour)
Multi-phase: Split the eruption into warning → tremor → eruption → cooldown phases with separate timers