🌀 Moirai

A universal orchestrator for dynamic, modular simulations.

Moirai is a lightweight, extensible Python framework for orchestrating heterogeneous simulations. A central Simulation runtime manages pluggable Module components — each responsible for its own domain logic.

Use it for: agent-based models, ECS systems, weather simulation, economic models, epidemiology, social dynamics, city simulators, power grids, supply chains, and any other dynamic system you can imagine.

---

✨ Philosophy

Simulation  →  universal runtime
Module      →  pluggable domain logic
EventBus    →  loose coupling between modules
Metrics     →  observable simulation state
Resources   →  shared data between modules
Clock       →  real-world time mapping
Scheduler   →  deferred & recurring callbacks
Snapshot    →  serialize state to JSON

---

🚀 Quick Start

Installation

pip install moirai-sim

Or from source:

git clone https://github.com/yourname/moirai
cd moirai
pip install -e ".[dev]"

Basic Usage

from moirai import Simulation
from moirai.clock import Clock
from moirai.examples.weather import WeatherModule
from moirai.examples.population import PopulationModule

sim = Simulation(seed=42, clock=Clock(time_step=1.0, unit="day"))

sim.attach(WeatherModule(initial_temp=15.0))
sim.attach(PopulationModule(initial_mood=100.0))

sim.run(365)

print(sim.summary())

---

🧩 Writing a Module

from moirai.module import Module

class EconomyModule(Module):
    phase    = "economy"   # execution phase (lexicographic order)
    priority = 0           # within-phase ordering (lower = earlier)
    enabled  = True        # set False to skip update() each tick

    def setup(self, sim) -> None:
        sim.resources.set("gdp", 1_000_000.0)

    def update(self, sim) -> None:
        growth = sim.rng.uniform(0.98, 1.03)
        gdp    = sim.resources.get("gdp")
        sim.resources.set("gdp", gdp * growth)
        sim.metrics.track("gdp", sim.resources.get("gdp"))
        sim.eventbus.emit("gdp_updated", value=sim.resources.get("gdp"))

    def teardown(self, sim) -> None:
        pass

---

🔌 Event Bus

# Standard subscription
sim.eventbus.subscribe("gdp_updated", lambda value, **_: print(f"GDP: {value:,.0f}"))

# One-shot (auto-unsubscribes after first call)
sim.eventbus.once("run_start", lambda **_: print("Simulation started!"))

# Wildcard — receives every event
sim.eventbus.subscribe("*", lambda **kw: log.debug("event received"))

# Decorator shorthand on the simulation
@sim.on("temperature_changed")
def log_temp(value, **_):
    print(f"[{sim.clock.format_time(sim.ticks)}] Temp: {value:.1f} °C")

---

⏱ Clock

from datetime import datetime
from moirai.clock import Clock

# 1 tick = 1 day, anchored to a calendar date
clock = Clock(time_step=1.0, unit="day", start_date=datetime(2025, 1, 1))

sim = Simulation(seed=42, clock=clock)
sim.run(90)

print(sim.clock.format_time(sim.ticks))   # "2025-04-01"
print(sim.time)                           # 90.0
print(clock.ticks_for(365))              # 365

---

🗓 Scheduler

# One-shot callback at tick 10
sim.scheduler.schedule_at(10, lambda: print("Tick 10!"))

# Recurring every 7 ticks (weekly event)
handle = sim.scheduler.schedule_every(7, lambda: print("Weekly checkpoint"))

# Cancel
sim.scheduler.cancel(handle)

---

📊 Metrics

# Track values
sim.metrics.track("population", 5_000)

# Query
print(sim.metrics.history("population"))  # full history
print(sim.metrics.latest("population"))   # most recent value

# Statistical summary
s = sim.metrics.summary("population")
print(s.mean, s.std, s.min, s.max, s.count)

# Export to JSON
sim.metrics.export_json("results/metrics.json")

# pandas-compatible dict
import pandas as pd
df = pd.DataFrame(sim.metrics.to_frame())

---

🪣 Resources

# Set, get, update
sim.resources.set("gdp", 1_000_000.0)
sim.resources.update({"inflation": 0.02, "unemployment": 0.05})

# Type-safe get — raises ResourceTypeError on mismatch
gdp = sim.resources.get_typed("gdp", float)

# set-if-absent
sim.resources.setdefault("mortality_rate", 0.01)

---

⏸ Pause & Resume

class CheckpointModule(Module):
    phase = "control"

    def setup(self, sim): pass
    def teardown(self, sim): pass

    def update(self, sim):
        if sim.ticks == 100:
            sim.pause()   # stops the loop after this tick

sim.attach(CheckpointModule())
ticks_run = sim.run(365)   # returns early at tick 100
print(sim.state)           # SimulationState.PAUSED
sim.run(265)               # resume for remaining ticks

---

📸 Snapshots

sim.run(100)

# Save
snap = sim.snapshot()
snap.save("checkpoint.json")

# Load & inspect
from moirai.snapshot import Snapshot
loaded = Snapshot.load("checkpoint.json")
print(loaded.tick)         # 100
print(loaded.resources)    # {"temperature": ...}
print(loaded.summary())

---

🔬 Built-in Examples

Module	Domain	Key resources
WeatherModule	Climate	temperature
PopulationModule	Social	population_mood
EpidemicModule	Epidemiology (SIR)	sir_S, sir_I, sir_R
EconomyModule	Macroeconomics	gdp, inflation, unemployment

from moirai.examples.epidemic import EpidemicModule

sim = Simulation(seed=0)
sim.eventbus.subscribe("epidemic_ended", lambda tick, **_: print(f"Ended at tick {tick}"))
sim.attach(EpidemicModule(population=100_000, initial_infected=10, beta=0.3, gamma=0.1))
sim.run(500)

---

📁 Project Structure

moirai/
├── pyproject.toml
├── README.md
├── CHANGELOG.md
├── src/
│   └── moirai/
│       ├── __init__.py          # Public API
│       ├── simulation.py        # Orchestrator + SimulationState
│       ├── module.py            # Abstract Module base
│       ├── eventbus.py          # Pub/sub (once, wildcard)
│       ├── metrics.py           # Time-series + MetricSummary
│       ├── resources.py         # Typed resource container
│       ├── clock.py             # Time-unit mapping
│       ├── scheduler.py         # Deferred/recurring callbacks
│       ├── snapshot.py          # JSON checkpoint
│       ├── exceptions.py        # Exception hierarchy
│       ├── py.typed             # PEP 561
│       └── examples/
│           ├── weather.py
│           ├── population.py
│           ├── epidemic.py      # SIR model
│           └── economy.py       # GDP / inflation / unemployment
└── tests/
    ├── conftest.py
    ├── test_simulation.py       # 68 tests
    ├── test_modules.py          # 85 tests
    ├── test_clock.py            # 17 tests
    ├── test_scheduler.py        # 20 tests
    └── test_snapshot.py         # 15 tests

---

🧪 Running Tests

pytest
pytest --cov=moirai --cov-report=term-missing

---

📄 License

MIT — see LICENSE for details.