Animation in SimPy: Making Simulations Come Alive

import simpy
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import numpy as np

class AnimatedSimulation:
    def __init__(self):
        self.time_data = []
        self.queue_data = []

    def run_step(self, env, server):
        """Run simulation and collect data."""
        self.time_data.append(env.now)
        self.queue_data.append(len(server.queue))

def animate_queue():
    fig, ax = plt.subplots()
    ax.set_xlim(0, 100)
    ax.set_ylim(0, 20)
    ax.set_xlabel('Time')
    ax.set_ylabel('Queue Length')

    line, = ax.plot([], [], lw=2)

    sim_data = AnimatedSimulation()

    def init():
        line.set_data([], [])
        return line,

    def update(frame):
        # Update simulation
        sim_data.time_data.append(frame)
        sim_data.queue_data.append(np.random.poisson(5))

        line.set_data(sim_data.time_data, sim_data.queue_data)
        return line,

    ani = FuncAnimation(fig, update, frames=range(100),
                       init_func=init, blit=True, interval=100)

    plt.show()

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

class RealTimeDisplay:
    def __init__(self):
        self.data_queue = queue.Queue()
        self.times = []
        self.values = []

    def update_data(self, time, value):
        self.data_queue.put((time, value))

    def run_animation(self):
        fig, ax = plt.subplots()
        ax.set_xlim(0, 100)
        ax.set_ylim(0, 20)
        line, = ax.plot([], [], lw=2)

        def update(frame):
            while not self.data_queue.empty():
                t, v = self.data_queue.get()
                self.times.append(t)
                self.values.append(v)
            line.set_data(self.times, self.values)
            if self.times:
                ax.set_xlim(0, max(self.times) + 10)
            return line,

        ani = FuncAnimation(fig, update, interval=100)
        plt.show()

# Run simulation in separate thread
def simulation_thread(display):
    env = simpy.rt.RealtimeEnvironment(factor=0.1)
    server = simpy.Resource(env, capacity=2)

    def monitor(env, server, display):
        while True:
            display.update_data(env.now, len(server.queue))
            yield env.timeout(1)

    env.process(monitor(env, server, display))
    # ... add other processes
    env.run(until=100)

display = RealTimeDisplay()
sim_thread = threading.Thread(target=simulation_thread, args=(display,))
sim_thread.start()
display.run_animation()

import time
import os

def clear_screen():
    os.system('cls' if os.name == 'nt' else 'clear')

def text_animation(env, server):
    while True:
        clear_screen()

        # Draw queue
        queue_len = len(server.queue)
        busy = server.count

        print(f"Time: {env.now:.1f}")
        print(f"Server: {'[BUSY]' * busy}{'[    ]' * (server.capacity - busy)}")
        print(f"Queue:  {'o' * queue_len}")
        print(f"\nWaiting: {queue_len}, In service: {busy}")

        yield env.timeout(0.5)

# Use with RealtimeEnvironment
env = simpy.rt.RealtimeEnvironment(factor=1)
server = simpy.Resource(env, capacity=2)
env.process(text_animation(env, server))

# Note: Requires pygame
import pygame
import simpy
import random

class PygameVisualisation:
    def __init__(self, width=800, height=600):
        pygame.init()
        self.screen = pygame.display.set_mode((width, height))
        self.clock = pygame.time.Clock()
        self.width = width
        self.height = height

        # Simulation state
        self.entities = []
        self.server_busy = False

    def draw(self):
        self.screen.fill((255, 255, 255))

        # Draw server
        color = (255, 0, 0) if self.server_busy else (0, 255, 0)
        pygame.draw.rect(self.screen, color, (350, 250, 100, 100))

        # Draw queue
        for i, entity in enumerate(self.entities):
            pygame.draw.circle(self.screen, (0, 0, 255),
                             (100 + i * 30, 300), 10)

        # Update display
        pygame.display.flip()

    def update(self, queue_length, server_busy):
        self.entities = list(range(queue_length))
        self.server_busy = server_busy

        # Handle events
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                pygame.quit()
                return False
        return True

# Integration with SimPy
def simulation_with_pygame(vis):
    env = simpy.rt.RealtimeEnvironment(factor=0.5)
    server = simpy.Resource(env, capacity=1)

    def display_update(env, server, vis):
        while True:
            if not vis.update(len(server.queue), server.count > 0):
                break
            vis.draw()
            yield env.timeout(0.1)

    env.process(display_update(env, server, vis))
    # ... add customers
    env.run(until=60)

import plotly.graph_objects as go
from plotly.subplots import make_subplots

def create_animated_plot(time_series):
    """Create animated plot with Plotly."""
    fig = go.Figure(
        data=[go.Scatter(x=[], y=[], mode='lines')],
        layout=go.Layout(
            xaxis=dict(range=[0, max(r['time'] for r in time_series)]),
            yaxis=dict(range=[0, max(r['queue'] for r in time_series) + 1]),
            title="Queue Length Animation",
            updatemenus=[dict(
                type="buttons",
                buttons=[dict(label="Play",
                             method="animate",
                             args=[None, {"frame": {"duration": 100}}])]
            )]
        ),
        frames=[
            go.Frame(data=[go.Scatter(
                x=[r['time'] for r in time_series[:i]],
                y=[r['queue'] for r in time_series[:i]]
            )])
            for i in range(1, len(time_series))
        ]
    )

    fig.write_html('animation.html')
    fig.show()

from matplotlib.animation import FuncAnimation, PillowWriter

def create_gif(time_series, filename='simulation.gif'):
    fig, ax = plt.subplots()
    times = [r['time'] for r in time_series]
    queues = [r['queue'] for r in time_series]

    ax.set_xlim(0, max(times))
    ax.set_ylim(0, max(queues) + 1)
    ax.set_xlabel('Time')
    ax.set_ylabel('Queue Length')

    line, = ax.plot([], [], lw=2)

    def update(frame):
        line.set_data(times[:frame], queues[:frame])
        return line,

    ani = FuncAnimation(fig, update, frames=len(times), interval=50)
    ani.save(filename, writer=PillowWriter(fps=20))
    print(f"Saved animation to {filename}")

# Note: Requires dash
from dash import Dash, dcc, html
from dash.dependencies import Input, Output
import plotly.graph_objs as go

app = Dash(__name__)

# Shared data
simulation_data = {'time': [], 'queue': []}

app.layout = html.Div([
    dcc.Graph(id='live-graph'),
    dcc.Interval(id='interval', interval=500)  # Update every 500ms
])

@app.callback(Output('live-graph', 'figure'),
              Input('interval', 'n_intervals'))
def update_graph(n):
    return {
        'data': [go.Scatter(
            x=simulation_data['time'],
            y=simulation_data['queue'],
            mode='lines'
        )],
        'layout': go.Layout(
            title='Live Queue Length',
            xaxis={'title': 'Time'},
            yaxis={'title': 'Queue'}
        )
    }

# Run simulation in background, update simulation_data
# app.run_server(debug=True)