Miscellaneous
Ki Insurance
Download
Cite
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())