Sedar Olmez - hades-framework 1.0.2

Miscellaneous

Ki Insurance

Download

Cite

APA Click to copy
Insurance, K. hades-framework 1.0.2.

Chicago/Turabian Click to copy
Insurance, Ki. “Hades-Framework 1.0.2,” n.d.

MLA Click to copy
Insurance, Ki. Hades-Framework 1.0.2.

BibTeX Click to copy

@misc{ki-a,
  title = {hades-framework 1.0.2},
  author = {Insurance, Ki}
}

HADES (HADES Asynchronous Discrete-Event Simulation) is a small, user friendly framework for creating simulations in python!

Features

🎲🤖 Supports both Agent Based and Process Based models - how you model the entities in your simulation is up to you!
⚡ Async execution within a time-step - designed for working IO-bound workloads over the network (e.g. LLM APIs, db lookups, etc)
📈 Visualisation - websockets support to for building a custom frontend for your sim, matplotlib in a Jupyter notepad or simply outputting a mermaid diagram
🏷️ Pydantic style immutable events - give type hints and enforcement, making it clear what an event will contain
📦 Encapsulated simulated processes - processes or agents are encapsulated, keeping state manageable and making it possible to swap processes in and out
😊 User friendly - pattern matching on pydantic based events makes for an intuitive way to build simulations, while the separation of state helps avoid potential footguns!

import asyncio
from enum import Enum

from hades import Event, Hades, NotificationResponse, Process, RandomProcess, SimulationStarted

# Let's define our Events


class HeroLifeCycleStage(Enum):
    SAFE = 1
    IN_DANGER = 2
    DECEASED = 3


class LightningBoltThrown(Event):
    target_id: str


class StormCreated(Event):
    target_id: str


class OdysseusDied(Event):
    pass


class AthenaIntervened(Event):
    target_id: str


# And now define our Processes


class Zeus(Process):
    async def notify(self, event: Event):
        match event:
            case SimulationStarted(t=t):
                for i in range(0, 100, 25):
                    self.add_event(LightningBoltThrown(t=t + i + 2, target_id="Odysseus"))
                return NotificationResponse.ACK
        return NotificationResponse.NO_ACK


class Poseidon(Process):
    async def notify(self, event: Event):
        match event:
            case SimulationStarted(t=t):
                for i in range(0, 100, 5):
                    self.add_event(StormCreated(t=t + i + 2, target_id="Odysseus"))
                return NotificationResponse.ACK
        return NotificationResponse.NO_ACK


class GoddessAthena(RandomProcess):
    async def notify(self, event: Event):
        match event:
            case SimulationStarted(t=t):
                self.add_event(AthenaIntervened(t=t + 3, target_id="Odysseus"))
                return NotificationResponse.ACK
            case OdysseusDied(t=t):
                if self.random.random() > 0.5:
                    self.add_event(AthenaIntervened(t=t, target_id="Odysseus"))
                else:
                    print("athena was too late to save odysseus")
                return NotificationResponse.ACK
        return NotificationResponse.NO_ACK


class Odysseus(RandomProcess):
    def __init__(self, seed):
        super().__init__(seed)
        self.status = HeroLifeCycleStage.SAFE
        self._health = 100

    @property
    def instance_identifier(self) -> str:
        return "Odysseus"

    def _handle_peril(self, t: int, max_damage: int, source: str):
        self.status = HeroLifeCycleStage.IN_DANGER
        print(f"odysseus is in danger from {source}!")
        lost_hp = round(self.random.random() * max_damage)
        self._health = max(self._health - lost_hp, 0)
        print(f"odysseus' health dropped to {self._health}")
        if self._health == 0:
            print("odysseus died")
            self.status = HeroLifeCycleStage.DECEASED
            self.add_event(OdysseusDied(t=t))

    async def notify(self, event: Event):
        match event:
            case LightningBoltThrown(t=t, target_id=target_id):
                if self.status == HeroLifeCycleStage.DECEASED:
                    return NotificationResponse.ACK_BUT_IGNORED
                self._handle_peril(t, 90, "Zeus' lightning bolt")
                return NotificationResponse.ACK
            case StormCreated(t=t, target_id=target_id):
                if target_id != self.instance_identifier or self.status == HeroLifeCycleStage.DECEASED:
                    return NotificationResponse.ACK_BUT_IGNORED
                self._handle_peril(t, 50, "Poseidon's storm")
                return NotificationResponse.ACK
            case AthenaIntervened(t=t, target_id=target_id):
                print("but athena intervened saving and healing odysseus to 100")
                self._health = 100
                self.status = HeroLifeCycleStage.SAFE
                return NotificationResponse.ACK
        return NotificationResponse.NO_ACK


# Finally lets compose them to build the simulation


async def odyssey():
    world = Hades()
    world.register_process(Zeus())
    world.register_process(Poseidon())
    world.register_process(Odysseus("pomegranate"))
    world.register_process(GoddessAthena("pomegranate"))

    await world.run()


if __name__ == "__main__":
    asyncio.run(odyssey())