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 typing import ClassVar

from maid_engine.core.ecs import Component, System
from maid_engine.core.world import World
from maid_stdlib.components import HealthComponent

BLOOD_MOON_MULTIPLIER = 1.5


class BloodMoonBuff(Component):
    """Marks a hostile as blood-moon-buffed and stores its TRUE base max HP.

    The baseline lives on a *persisted component* rather than a process-local
    dict. EntityDocument saves components, so if the server restarts while the
    blood moon is up, the real baseline survives. Re-firing the buff then sees
    the component already present and does NOT re-capture the (already boosted)
    maximum — which is exactly what would otherwise compound HP across restarts.
    Register it for persistence (see "Wiring into a Content Pack").
    """

    component_type: ClassVar[str] = "BloodMoonBuff"
    base_maximum: int = 0


async def blood_moon_rise(world: World) -> None:
    """Buff all hostile NPCs while the blood moon is up (idempotent + durable).

    The naive ``health.maximum = int(health.maximum * 1.5)`` multiplies the
    *current* maximum every cycle, compounding to ``base * 1.5**N``. A
    process-local baseline dict stops in-process compounding but is lost on
    restart, so a reboot mid-event would recapture the already-boosted maximum
    and keep growing. Anchoring the baseline to a **persisted** ``BloodMoonBuff``
    component makes repeated calls idempotent *and* restart-safe.
    """
    for entity in world.entities.with_tag("hostile"):
        health = entity.try_get(HealthComponent)
        if not health:
            continue
        buff = entity.try_get(BloodMoonBuff)
        if buff is None:
            # First application: capture the true baseline durably.
            entity.add(BloodMoonBuff(base_maximum=health.maximum))
            base = health.maximum
        else:
            # Already buffed (possibly from before a restart): reuse the stored
            # baseline; never recapture the current (boosted) maximum.
            base = buff.base_maximum
        health.maximum = int(base * BLOOD_MOON_MULTIPLIER)
        if health.current > health.maximum:
            health.current = health.maximum


async def blood_moon_set(world: World) -> None:
    """Restore hostiles to their baseline maximum when the blood moon ends."""
    for entity in world.entities.with_tag("hostile"):
        buff = entity.try_get(BloodMoonBuff)
        if buff is None:
            continue
        health = entity.try_get(HealthComponent)
        if health:
            health.maximum = buff.base_maximum
            if health.current > health.maximum:
                health.current = health.maximum
        entity.remove(BloodMoonBuff)


# In get_systems(): the buff is idempotent, so re-firing it is safe. For a true
# temporary event, schedule blood_moon_set after a duration to revert the buff.
systems = [
    RecurringTimerSystem(
        world,
        interval=1800.0,  # Every 30 minutes
        callback=partial(blood_moon_rise, 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.persistence.registry import ComponentRegistry
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 register_component_types(self, registry: ComponentRegistry) -> None:
        # Persist the blood-moon baseline so the buff can't compound or get
        # stranded across a restart (see blood_moon_rise above).
        registry.register(BloodMoonBuff, pack_name="my-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

1. System.update(delta) is called every tick with the elapsed time in seconds
2. The system accumulates delta in a timer and fires when the threshold is reached
3. Emitting VolcanoEruptionEvent lets other systems react (e.g., update room descriptions, play sounds)
4. world.entities_in_room() efficiently queries all entities in the blast radius
5. 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

See Also

• ECS Systems — System priority and tick lifecycle
• Events Guide — Emitting and subscribing to events
• Weather Effects — Another time-varying world system