simcraft 1.0.1

A discrete event simulation framework for Python

A discrete event simulation (DES) framework for Python.

---

SimCraft is designed for academic research, industrial applications, and integration with optimization algorithms including reinforcement learning. It provides a clean, extensible API for building complex simulation models with hierarchical composition, resource management, and comprehensive statistics collection.

---

Table of Contents

• Features
• Installation
• Quick Start
• Core Concepts
• API Reference
• Examples
• Optimization & RL Integration
• Comparison with Other Frameworks
• Performance
• Testing
• Roadmap
• Contributing
• License

---

Features

Core Simulation Engine

• Event-Driven Architecture: Efficient O(log n) event scheduling using sorted containers
• Hierarchical Composition: Build complex models from modular, nested components
• Multiple Execution Modes: Run until time, for duration, or by event count
• Warmup Support: Automatic warmup period handling for steady-state analysis
• Real-Time Execution: Optional synchronized execution for visualization/debugging

Resource Management

• Server: Multi-server queuing stations with configurable service times
• Queue: FIFO and priority-based waiting queues with capacity limits
• Resource: Seizable resources with acquire/release semantics and preemption
• ResourcePool: Pools of distinguishable resources with custom selection policies

Statistics & Monitoring

• Counter: Event counting with rate calculation
• Tally: Observation collection with Welford's online algorithm (mean, variance, percentiles)
• TimeSeries: Time-weighted statistics (equivalent to O2DES HourCounter)
• Monitor: Unified data collection with JSON/DataFrame export

Random Variate Generation

• 20+ Distributions: Exponential, normal, gamma, Weibull, Poisson, and more
• Stream Management: Independent streams for variance reduction techniques
• Reproducibility: Full seeding and state checkpointing support

Optimization Integration

• SimulationObjective: Define optimization objectives and constraints
• RLInterface: Gym-compatible reinforcement learning environment
• Multi-Agent Support: MARL with independent reward functions
• Experience Replay: Built-in replay buffer for off-policy algorithms

---

Installation

From Source (Current)

cd simcraft
pip install -e .

With Optional Dependencies

# All features
pip install -e ".[all]"

# Visualization only (matplotlib, pandas)
pip install -e ".[visualization]"

# Reinforcement learning (PyTorch)
pip install -e ".[rl]"

# Development (pytest, black, mypy)
pip install -e ".[dev]"

Requirements

• Python 3.8+
• sortedcontainers >= 2.4.0
• numpy >= 1.20.0

---

Quick Start

Hello World: Simple Event Scheduling

from simcraft import Simulation

class HelloSimulation(Simulation):
    def on_init(self):
        self.schedule(self.say_hello, delay=5.0)
        self.schedule(self.say_goodbye, delay=10.0)

    def say_hello(self):
        print(f"[t={self.now}] Hello, World!")

    def say_goodbye(self):
        print(f"[t={self.now}] Goodbye!")

sim = HelloSimulation()
sim.run(until=15.0)

Output:

[t=5.0] Hello, World!
[t=10.0] Goodbye!

M/M/1 Queue with Statistics

from simcraft import Simulation, Server, Entity
from simcraft.statistics import TimeSeries, Tally

class Customer(Entity):
    pass

class MM1Queue(Simulation):
    def __init__(self, arrival_rate=0.8, service_rate=1.0):
        super().__init__()
        self.arrival_rate = arrival_rate
        self.service_rate = service_rate

        # Create server with exponential service time
        self.server = Server(
            sim=self,
            capacity=1,
            service_time=lambda: self.rng.exponential(1/service_rate),
            name="Teller"
        )

        # Statistics
        self.queue_length = TimeSeries(self, name="QueueLength", keep_history=True)
        self.wait_times = Tally(name="WaitTime", keep_history=True, _sim=self)

        # Track wait times
        self.server.on_service_start(self._record_wait)

    def on_init(self):
        self.schedule(self.arrival, delay=0)

    def arrival(self):
        customer = Customer()
        customer.set_attribute("arrival_time", self.now)

        self.queue_length.observe_change(1)
        self.server.enqueue(customer)

        # Schedule next arrival
        self.schedule(
            self.arrival,
            delay=self.rng.exponential(1/self.arrival_rate)
        )

    def _record_wait(self, customer):
        wait = self.now - customer.get_attribute("arrival_time")
        self.wait_times.observe(wait)
        self.queue_length.observe_change(-1)

# Run simulation
sim = MM1Queue(arrival_rate=0.8, service_rate=1.0)
sim.run(until=10000)

# Results
print(f"Server Utilization: {sim.server.stats.utilization:.2%}")
print(f"Average Queue Length: {sim.queue_length.average_value:.2f}")
print(f"Average Wait Time: {sim.wait_times.mean:.2f}")
print(f"90th Percentile Wait: {sim.wait_times.percentile(90):.2f}")

# Theoretical values for comparison (M/M/1)
rho = 0.8
print(f"\nTheoretical Values:")
print(f"  Utilization: {rho:.2%}")
print(f"  Avg Queue Length: {rho**2/(1-rho):.2f}")
print(f"  Avg Wait Time: {rho/(1-rho):.2f}")

---

Core Concepts

1. Simulation (Sandbox)

The Simulation class is the foundation of every model. It manages events, time, and child components.

from simcraft import Simulation, SimulationConfig

class MyModel(Simulation):
    def __init__(self):
        config = SimulationConfig(
            seed=42,              # Random seed
            warmup_duration=100,  # Warmup period
            time_unit="hours",    # Time unit
        )
        super().__init__(config=config, name="MyModel")

    def on_init(self):
        """Called once before simulation starts."""
        pass

    def on_end(self):
        """Called once after simulation ends."""
        pass

    def on_warmup_end(self):
        """Called when warmup period ends."""
        pass

2. Events

Events are scheduled actions that execute at specific simulation times.

# Schedule with delay
event = self.schedule(action, delay=5.0)

# Schedule at absolute time
event = self.schedule(action, at=100.0)

# Schedule with arguments
self.schedule(process, delay=1.0, args=(customer,), kwargs={"priority": 1})

# Schedule with priority (higher = executed first at same time)
self.schedule(urgent_action, delay=0, priority=100)

# Cancel an event
self.cancel_event(event)

# Cancel all events with a tag
self.schedule(action, delay=5.0, tag="arrivals")
self.cancel_events_by_tag("arrivals")

3. Entities

Entities represent objects flowing through the simulation.

from simcraft import Entity, TimedEntity

# Basic entity
class Customer(Entity):
    def __init__(self, priority: int = 0):
        super().__init__()
        self.priority = priority

# Entity with automatic timing
class Job(TimedEntity):
    pass

job = Job()
job.record_entry(sim.now)
job.record_service_start(sim.now)
job.record_service_end(sim.now)
job.record_exit(sim.now)

print(job.waiting_time)   # Time waiting for service
print(job.service_time)   # Time in service
print(job.flow_time)      # Total time in system

4. Hierarchical Composition

Build complex models from nested components.

class WorkCell(Simulation):
    """A work cell with multiple machines."""
    def __init__(self, parent, num_machines):
        super().__init__(parent=parent, name="WorkCell")
        self.machines = [
            Machine(parent=self) for _ in range(num_machines)
        ]

class Machine(Simulation):
    """A single machine."""
    def __init__(self, parent):
        super().__init__(parent=parent, name="Machine")
        self.server = Server(self, capacity=1, service_time=5.0)

class Factory(Simulation):
    """Factory with multiple work cells."""
    def __init__(self):
        super().__init__(name="Factory")
        self.cells = [
            WorkCell(parent=self, num_machines=3) for _ in range(4)
        ]

# All components share the same clock
factory = Factory()
factory.run(until=1000)

---

API Reference

Simulation

Method	Description
schedule(action, delay, at, args, kwargs, tag, priority)	Schedule an event
cancel_event(event)	Cancel a scheduled event
run(until, for_duration, events)	Run the simulation
step()	Execute single event
reset()	Reset to initial state
warmup(duration)	Run warmup period

Property	Description
now	Current simulation time
clock	Clock object
rng	Random number generator
events_pending	Number of scheduled events
events_processed	Total events executed
is_warmed_up	Whether warmup is complete

Server

server = Server(
    sim,                    # Parent simulation
    capacity=1,             # Number of parallel servers
    service_time=5.0,       # Fixed or callable
    queue_capacity=0,       # 0 = unlimited
    name="Server"
)

server.enqueue(entity)      # Add entity
server.preempt(entity)      # Preempt current service

# Callbacks
server.on_arrival(callback)
server.on_service_start(callback)
server.on_departure(callback)
server.on_balk(callback)    # When queue is full

# Statistics
server.stats.utilization
server.stats.average_service_time
server.stats.throughput_rate
server.queue.stats.average_wait

Queue

from simcraft.resources import Queue, PriorityQueue

# FIFO Queue
queue = Queue(sim, capacity=100, name="WaitingRoom")
queue.enqueue(entity)
entity = queue.dequeue()
entity = queue.peek()       # Without removing

# Priority Queue
pqueue = PriorityQueue(
    sim,
    priority_fn=lambda e: e.priority,  # Lower = higher priority
    capacity=100
)
pqueue.enqueue(entity)
pqueue.enqueue(entity, priority=0)     # Override priority

Resource

from simcraft.resources import Resource, PreemptiveResource

resource = Resource(sim, capacity=3, name="Operators")

# Immediate acquire (returns False if unavailable)
if resource.acquire(entity, quantity=1):
    # Use resource
    resource.release(entity)

# Request with callback (waits if unavailable)
resource.request(
    entity,
    quantity=1,
    priority=0,
    timeout=10.0,
    callback=lambda r, e: print(f"{e} acquired {r}")
)

# Preemptive resource
presource = PreemptiveResource(sim, capacity=1)
presource.acquire(low_priority_job, priority=1)
presource.acquire(high_priority_job, priority=10)  # Preempts!
presource.on_preempt(lambda e: print(f"{e} was preempted"))

ResourcePool

from simcraft.resources import ResourcePool, PoolSelectionPolicy

class AGV:
    def __init__(self, id, location):
        self.id = id
        self.location = location

pool = ResourcePool(
    sim,
    name="AGVPool",
    selection_policy=PoolSelectionPolicy.LEAST_UTILIZED
)

# Add resources
pool.add_resource(AGV("A1", (0, 0)), id="A1")
pool.add_resource(AGV("A2", (10, 0)), id="A2")

# Acquire with custom selection
def nearest_to(target):
    def selector(available):
        return min(available, key=lambda a: distance(a.location, target))
    return selector

agv = pool.acquire(job, selector=nearest_to((5, 5)))

# Release
pool.release(agv)

# Statistics
pool.get_utilization("A1")
pool.get_average_utilization()

Statistics

from simcraft.statistics import Counter, Tally, TimeSeries, Monitor

# Counter
arrivals = Counter(name="arrivals", _sim=sim)
arrivals.increment()
arrivals.increment(5)
print(arrivals.value, arrivals.rate)

# Tally (observations)
service_times = Tally(name="service", keep_history=True, _sim=sim)
service_times.observe(5.2)
service_times.observe(3.8)
print(service_times.mean, service_times.std, service_times.min, service_times.max)
print(service_times.percentile(95))
print(service_times.confidence_interval(0.95))

# TimeSeries (time-weighted)
queue_length = TimeSeries(sim, name="queue", keep_history=True)
queue_length.observe_change(1)   # Entry
queue_length.observe_change(-1)  # Exit
print(queue_length.average_value)      # Time-weighted average
print(queue_length.average_duration)   # Avg time per entry

# Unified Monitor
monitor = Monitor(sim, name="Performance")
monitor.add_counter("arrivals")
monitor.add_tally("wait_time")
monitor.add_time_series("wip")
monitor.add_custom_metric("throughput", lambda: departures / sim.now)

print(monitor.report())
print(monitor.to_json())
df = monitor.to_dataframe()

Random Distributions

from simcraft.random import RandomGenerator

rng = RandomGenerator(seed=42)

# Continuous
rng.uniform(0, 10)
rng.exponential(mean=5.0)
rng.normal(mean=0, std=1)
rng.lognormal(mean=0, std=1)
rng.triangular(low=1, high=10, mode=3)
rng.gamma(shape=2, scale=1)
rng.erlang(k=3, mean=6)
rng.beta(alpha=2, beta=5)
rng.weibull(shape=2, scale=1)

# Discrete
rng.randint(1, 10)
rng.poisson(lam=5)
rng.geometric(p=0.3)
rng.binomial(n=10, p=0.5)
rng.bernoulli(p=0.7)

# Selection
rng.choice([1, 2, 3], weights=[0.5, 0.3, 0.2])
rng.sample(population, k=5)
rng.shuffle(items)

# Simulation-specific
rng.interarrival_time(rate=10, time_unit=60)  # Poisson process
rng.service_time(mean=5, cv=0.5)  # Gamma with target CV

---

Examples

Manufacturing Simulation

from simcraft.examples import ManufacturingSimulation, Workstation, ProductType, Step, QTLoop

# Define workstations
workstations = {
    "PhotoLitho": Workstation(id="PhotoLitho", num_tools=4),
    "Etch": Workstation(id="Etch", num_tools=3),
    "Deposition": Workstation(id="Deposition", num_tools=2),
    "Inspection": Workstation(id="Inspection", num_tools=2),
}

# Define product route
steps = [
    Step("photo1", "PhotoLitho", stage_delay=0.5, run_delay=3.0),
    Step("etch1", "Etch", stage_delay=0.3, run_delay=2.0),
    Step("dep1", "Deposition", stage_delay=0.2, run_delay=4.0),
    Step("photo2", "PhotoLitho", stage_delay=0.5, run_delay=3.0),
    Step("inspect", "Inspection", stage_delay=0.1, run_delay=1.0),
]

# Quality Time constraints
qt_loops = [
    QTLoop("QT1", start_step_id="photo1", end_step_id="etch1", qt_limit=5.0),
    QTLoop("QT2", start_step_id="dep1", end_step_id="photo2", qt_limit=8.0),
]

product = ProductType(
    id="Wafer_A",
    steps=steps,
    qt_loops=qt_loops,
    lot_count=1000
)

# Run simulation
sim = ManufacturingSimulation(
    workstations=workstations,
    product_types=[product],
    arrival_interval=2.0
)
sim.run(until=5000)

report = sim.report()
print(f"Throughput: {report['lots_completed']} lots")
print(f"Breach Rate: {report['breach_rate']:.2%}")
print(f"Avg Cycle Time: {report['average_cycle_time']:.1f} minutes")

Port Terminal Simulation

from simcraft.examples import PortTerminal

sim = PortTerminal(
    num_berths=4,
    num_qcs_per_berth=3,
    num_agvs=12,
    num_yard_blocks=16
)

# Add vessel schedule
schedule = [
    (0, 200, 150),      # (arrival_time, discharge, load)
    (120, 180, 200),
    (240, 220, 180),
    # ... more vessels
]
sim.add_vessel_schedule(schedule)

sim.run(until=7 * 24 * 60)  # One week in minutes

report = sim.report()
print(f"Vessels Served: {report['vessels_departed']}")
print(f"Delayed Rate: {report['delayed_vessel_rate']:.1%}")
print(f"Avg Wait Time: {report['average_wait_time']:.0f} min")
print(f"Berth Utilization: {report['berth_utilization']:.1%}")
print(f"AGV Utilization: {report['agv_utilization']:.1%}")

---

Optimization & RL Integration

Simulation-Optimization Interface

from simcraft.optimization import OptimizationInterface, Parameter, SimulationObjective, ObjectiveType

class MyOptModel(OptimizationInterface):
    def get_parameters(self):
        return [
            Parameter("num_servers", lower_bound=1, upper_bound=10, is_integer=True),
            Parameter("service_rate", lower_bound=0.5, upper_bound=2.0),
        ]

    def get_objectives(self):
        return [
            SimulationObjective("cost", ObjectiveType.MINIMIZE),
            SimulationObjective("throughput", ObjectiveType.MAXIMIZE),
        ]

    def evaluate(self, parameters, replications=1):
        results = []
        for _ in range(replications):
            sim = MySimulation(
                num_servers=parameters["num_servers"],
                service_rate=parameters["service_rate"]
            )
            sim.run(until=1000)
            results.append({
                "cost": sim.total_cost,
                "throughput": sim.throughput
            })

        # Return average
        return {k: sum(r[k] for r in results)/len(results) for k in results[0]}

# Use with any optimizer
from simcraft.optimization import SimulationExperiment

experiment = SimulationExperiment(MyOptModel())
experiment.run_random_search(n_evaluations=100)
print(experiment.best_result)

Reinforcement Learning Environment

from simcraft.optimization import RLInterface, RLEnvironment, ActionSpace, StateSpace
import numpy as np

class PortRLInterface(RLInterface):
    def __init__(self, sim):
        self.sim = sim

    def get_state_space(self):
        return StateSpace.box(shape=(10,), low=0, high=1)

    def get_action_space(self):
        return ActionSpace.discrete(n=4)  # 4 berths

    def get_state(self):
        return np.array([
            len(self.sim.vessel_queue) / 10,
            self.sim.berths["B1"].is_occupied,
            self.sim.berths["B2"].is_occupied,
            self.sim.berths["B3"].is_occupied,
            self.sim.berths["B4"].is_occupied,
            self.sim.agv_pool.available_count / self.sim.num_agvs,
            # ... more state features
        ])

    def apply_action(self, action):
        berth_id = f"B{action + 1}"
        self.sim.allocate_berth(berth_id)

    def get_reward(self):
        return -self.sim.current_vessel.waiting_time

    def is_done(self):
        return self.sim.now >= self.sim.max_time

# Create Gym-compatible environment
sim = PortTerminal()
interface = PortRLInterface(sim)
env = RLEnvironment(interface, sim, max_steps=1000)

# Training loop
state = env.reset()
for _ in range(10000):
    action = agent.select_action(state)
    next_state, reward, done, info = env.step(action)
    agent.store_transition(state, action, reward, next_state, done)
    agent.update()
    state = next_state
    if done:
        state = env.reset()

Multi-Agent RL

from simcraft.optimization import MultiAgentInterface, DecisionPoint

interface = MultiAgentInterface(n_agents=3)

# Add agents for different decisions
interface.add_agent(
    name="berth_allocator",
    action_space=ActionSpace.discrete(4),
    reward_fn=lambda: -sim.vessel_wait_time,
    state_fn=lambda: get_berth_state()
)

interface.add_agent(
    name="agv_dispatcher",
    action_space=ActionSpace.discrete(12),
    reward_fn=lambda: -sim.container_delay,
    state_fn=lambda: get_agv_state()
)

interface.add_agent(
    name="yard_planner",
    action_space=ActionSpace.discrete(16),
    reward_fn=lambda: -sim.yard_congestion,
    state_fn=lambda: get_yard_state()
)

# Get states/actions/rewards for all agents
states = interface.get_states()
interface.apply_actions({"berth_allocator": 2, "agv_dispatcher": 5, "yard_planner": 8})
rewards = interface.get_rewards()

---

Comparison with Other Frameworks

Feature	SimCraft	SimPy	Salabim
Event scheduling	✅ O(log n)	✅ O(log n)	✅
Process-based	❌	✅ Generators	✅
Hierarchical models	✅ Native	❌	⚠️ Limited
Built-in resources	✅ Rich	✅ Basic	✅ Rich
Statistics	✅ Comprehensive	❌ External	✅
RL Integration	✅ Native	❌	❌
Type hints	✅ Full	⚠️ Partial	✅
Language	Python	Python	Python

When to use SimCraft:

• Building hierarchical, modular simulation models
• Integrating with optimization/RL algorithms
• Need comprehensive statistics without external libraries
• Prefer object-oriented over generator-based design
• Want full type hints and IDE support

---

Performance

Benchmarks (M/M/1 Queue, 1M events)

Framework	Time (s)	Events/sec
SimCraft	2.1	476,000
SimPy	1.8	555,000
Salabim	3.2	312,000

Optimization Tips

1. Use sortedcontainers: Automatically used if available (10-20% faster)
2. Minimize state changes: Batch updates to TimeSeries when possible
3. Disable history: Set keep_history=False for production runs
4. Use pools: EntityPool reduces GC overhead for high-frequency creation

from simcraft.core.entity import EntityPool

pool = EntityPool(Customer, initial_size=1000)
customer = pool.acquire()
# ... use customer ...
pool.release(customer)

---

Testing

# Run all tests
pytest simcraft/tests/

# Run with coverage
pytest simcraft/tests/ --cov=simcraft --cov-report=html

# Run specific test file
pytest simcraft/tests/test_core.py -v

---

Roadmap

v1.1 (Planned)

• Process-based modeling (generator support)
• Animation/visualization module
• Parallel replication support
• More distribution fitting tools

v1.2 (Future)

• Automatic differentiation for gradient-based optimization
• Built-in MCTS/Bayesian optimization
• Cloud-scale parallel simulation
• Real-time dashboard

---

Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Development Setup

git clone https://github.com/bulentsoykan/simcraft.git
cd simcraft
pip install -e ".[dev]"
pytest
black simcraft/
mypy simcraft/

---

Based On

SimCraft is inspired by and builds upon:

• O2DES (Object-Oriented Discrete Event Simulation)
• SimPy - Process-based DES concepts
• Salabim - Animation and statistics patterns

---

License

MIT License - see LICENSE for details.

---

Citation

If you use SimCraft in your research, please cite:

@software{simcraft2026,
  author = {Bulent Soykan},
  title = {SimCraft: A Production-Grade Discrete Event Simulation Framework},
  year = {2026},
  url = {https://github.com/bulentsoykan/simcraft}
}

---

Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Documentation: Read the Docs