Normal SimPy runs as fast as possible. Simulation years pass in seconds. But sometimes you need simulation time to match real time.

When to Use Real-Time Simulation

• Live demonstrations - Show stakeholders the simulation running
• Hardware-in-the-loop - Interface with real equipment
• Training systems - Operators interact in real time
• Testing - Validate timing-sensitive code

RealtimeEnvironment

import simpy.rt

env = simpy.rt.RealtimeEnvironment(factor=1.0)

def process(env):
    print(f"Start at real time")
    yield env.timeout(5)  # Actually waits 5 seconds
    print(f"5 seconds have passed")

env.process(process(env))
env.run(until=10)  # Takes 10 real seconds

Speed Factor

Control how fast simulation runs relative to real time:

# Real-time (1 sim second = 1 real second)
env = simpy.rt.RealtimeEnvironment(factor=1.0)

# Half speed (1 sim second = 2 real seconds)
env = simpy.rt.RealtimeEnvironment(factor=0.5)

# Double speed (1 sim second = 0.5 real seconds)
env = simpy.rt.RealtimeEnvironment(factor=2.0)

# 10x speed (1 sim second = 0.1 real seconds)
env = simpy.rt.RealtimeEnvironment(factor=10.0)

Strict Mode

By default, SimPy tries to keep up with real time but doesn’t guarantee it:

# Non-strict (default): proceeds even if behind schedule
env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)

# Strict: raises error if simulation can't keep up
env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=True)

Use strict mode when timing accuracy is critical.

Practical Example: Traffic Light

import simpy.rt
import time

def traffic_light(env):
    while True:
        print(f"[{time.strftime('%H:%M:%S')}] GREEN")
        yield env.timeout(30)

        print(f"[{time.strftime('%H:%M:%S')}] YELLOW")
        yield env.timeout(5)

        print(f"[{time.strftime('%H:%M:%S')}] RED")
        yield env.timeout(30)

env = simpy.rt.RealtimeEnvironment(factor=1.0)
env.process(traffic_light(env))
env.run(until=120)  # Run for 2 minutes real time

Interactive Simulation

Accept user input during simulation:

import simpy.rt
import threading
import queue

command_queue = queue.Queue()

def user_input_thread():
    """Background thread for user input."""
    while True:
        cmd = input("Enter command (add/quit): ")
        command_queue.put(cmd)
        if cmd == 'quit':
            break

def command_processor(env, server):
    """Process user commands."""
    while True:
        yield env.timeout(0.1)  # Check every 0.1 sim seconds
        try:
            cmd = command_queue.get_nowait()
            if cmd == 'add':
                env.process(customer(env, f"Manual-{env.now:.1f}", server))
                print(f"Added customer at {env.now:.1f}")
            elif cmd == 'quit':
                return
        except queue.Empty:
            pass

# Start input thread
input_thread = threading.Thread(target=user_input_thread, daemon=True)
input_thread.start()

env = simpy.rt.RealtimeEnvironment(factor=1.0)
server = simpy.Resource(env, capacity=1)
env.process(command_processor(env, server))
env.run(until=300)

Syncing with External Systems

Interface with real hardware or APIs:

import simpy.rt
import requests

def external_data_fetcher(env, url, interval):
    """Fetch data from external API at regular intervals."""
    while True:
        yield env.timeout(interval)
        response = requests.get(url)
        data = response.json()
        print(f"[{env.now}] Fetched: {data}")
        # Process data, update simulation state, etc.

env = simpy.rt.RealtimeEnvironment(factor=1.0)
env.process(external_data_fetcher(env, 'http://api.example.com/data', 5))
env.run(until=60)

Visualisation in Real Time

Update a display as simulation runs:

import simpy.rt
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation

queue_lengths = []
times = []

def monitor(env, server):
    global queue_lengths, times
    while True:
        times.append(env.now)
        queue_lengths.append(len(server.queue))
        yield env.timeout(0.5)

def animate(frame):
    plt.cla()
    plt.plot(times, queue_lengths)
    plt.xlabel('Time')
    plt.ylabel('Queue Length')
    plt.title('Real-Time Queue Monitor')

# Setup
env = simpy.rt.RealtimeEnvironment(factor=1.0)
server = simpy.Resource(env, capacity=1)
env.process(arrivals(env, server))
env.process(monitor(env, server))

# Animation
fig = plt.figure()
ani = FuncAnimation(fig, animate, interval=500)  # Update every 500ms

# Run simulation in background thread
import threading
sim_thread = threading.Thread(target=lambda: env.run(until=60))
sim_thread.start()

plt.show()

Logging with Real Timestamps

import simpy.rt
import logging
import time

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | sim_t=%(sim_time)s | %(message)s'
)

class SimTimeFilter(logging.Filter):
    def __init__(self, env):
        self.env = env

    def filter(self, record):
        record.sim_time = f"{self.env.now:.2f}"
        return True

env = simpy.rt.RealtimeEnvironment(factor=1.0)
logger = logging.getLogger()
logger.addFilter(SimTimeFilter(env))

def process(env):
    logger.info("Process started")
    yield env.timeout(5)
    logger.info("Process completed")

Performance Considerations

Real-time simulation has constraints:

# Bad: Complex computation blocks real-time sync
def slow_process(env):
    yield env.timeout(1)
    result = very_slow_computation()  # Takes 3 real seconds
    # Now we're behind schedule!

# Better: Offload heavy computation
import concurrent.futures

executor = concurrent.futures.ThreadPoolExecutor()

def async_process(env):
    yield env.timeout(1)
    future = executor.submit(very_slow_computation)
    # Continue simulation while computation runs

Switching Between Modes

Same model, different execution:

def create_simulation(realtime=False, factor=1.0):
    if realtime:
        env = simpy.rt.RealtimeEnvironment(factor=factor)
    else:
        env = simpy.Environment()

    # Setup simulation (same code either way)
    server = simpy.Resource(env, capacity=2)
    env.process(arrivals(env, server))

    return env

# Fast simulation for analysis
env = create_simulation(realtime=False)
env.run(until=10000)

# Real-time for demonstration
env = create_simulation(realtime=True, factor=1.0)
env.run(until=60)

Gotchas

Simulation Falling Behind

# With strict=True, this raises an error if can't keep up
env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=True)

# With strict=False, simulation proceeds but timing drifts
env = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)

Blocking Operations

# Don't do this - blocks the simulation
def bad_process(env):
    import time
    time.sleep(5)  # Blocks everything!
    yield env.timeout(0)

# Use simulation timeout instead
def good_process(env):
    yield env.timeout(5)  # Proper simulation delay

Summary

Real-time simulation:

• Use simpy.rt.RealtimeEnvironment
• Control speed with factor
• Use strict=True when timing matters
• Avoid blocking operations
• Good for demos, testing, hardware integration

Sometimes you need to slow down to see what’s happening.

Next Steps

• SimPy Environment Explained
• Visualising Results
• Logging in SimPy