Real-Time Simulation with SimPy: Syncing with the Real World

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

# 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)

# 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)

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

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)

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)

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()

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")

# 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

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)

# 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)

# 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