SimPy模拟框架Skill simpy

---

名称: simpy 描述: 基于Python的过程离散事件模拟框架。当构建涉及过程、队列、资源和基于时间事件的系统模拟时使用此技能，例如制造系统、服务操作、网络流量、物流，或任何实体随时间与共享资源交互的系统。

SimPy - 离散事件模拟

概述

SimPy是一个基于标准Python的过程离散事件模拟框架。使用SimPy建模系统，其中实体（客户、车辆、数据包等）随时间相互交互并竞争共享资源（服务器、机器、带宽等）。

核心能力:

• 使用Python生成器函数的过程建模
• 共享资源管理（服务器、容器、存储）
• 事件驱动调度和同步
• 与挂钟时间同步的实时模拟
• 全面的监控和数据收集

何时使用此技能

在以下情况使用SimPy技能：

1. 建模离散事件系统 - 事件在不规则间隔发生的系统
2. 资源竞争 - 实体竞争有限资源（服务器、机器、人员）
3. 队列分析 - 研究等待线、服务时间和吞吐量
4. 过程优化 - 分析制造、物流或服务过程
5. 网络模拟 - 数据包路由、带宽分配、延迟分析
6. 容量规划 - 确定期望性能的最佳资源水平
7. 系统验证 - 在实施前测试系统行为

不适用于:

• 具有固定时间步长的连续模拟（考虑SciPy ODE求解器）
• 没有资源共享的独立过程
• 纯数学优化（考虑SciPy优化）

快速开始

基本模拟结构

import simpy

def process(env, name):
    """一个简单的过程，等待并打印。"""
    print(f'{name} starting at {env.now}')
    yield env.timeout(5)
    print(f'{name} finishing at {env.now}')

# 创建环境
env = simpy.Environment()

# 启动过程
env.process(process(env, 'Process 1'))
env.process(process(env, 'Process 2'))

# 运行模拟
env.run(until=10)

资源使用模式

import simpy

def customer(env, name, resource):
    """客户请求资源，使用它，然后释放。"""
    with resource.request() as req:
        yield req  # 等待资源
        print(f'{name} got resource at {env.now}')
        yield env.timeout(3)  # 使用资源
        print(f'{name} released resource at {env.now}')

env = simpy.Environment()
server = simpy.Resource(env, capacity=1)

env.process(customer(env, 'Customer 1', server))
env.process(customer(env, 'Customer 2', server))
env.run()

核心概念

1. 环境

模拟环境管理时间和调度事件。

import simpy

# 标准环境（尽可能快运行）
env = simpy.Environment(initial_time=0)

# 实时环境（与挂钟同步）
import simpy.rt
env_rt = simpy.rt.RealtimeEnvironment(factor=1.0)

# 运行模拟
env.run(until=100)  # 运行到时间100
env.run()  # 运行直到没有事件剩余

2. 过程

过程使用Python生成器函数定义（带有yield语句的函数）。

def my_process(env, param1, param2):
    """产生事件以暂停执行的过程。"""
    print(f'Starting at {env.now}')

    # 等待时间流逝
    yield env.timeout(5)

    print(f'Resumed at {env.now}')

    # 等待另一个事件
    yield env.timeout(3)

    print(f'Done at {env.now}')
    return 'result'

# 启动过程
env.process(my_process(env, 'value1', 'value2'))

3. 事件

事件是过程同步的基本机制。过程产生事件，并在这些事件被触发时恢复。

常见事件类型:

• env.timeout(delay) - 等待时间流逝
• resource.request() - 请求资源
• env.event() - 创建自定义事件
• env.process(func()) - 过程作为事件
• event1 & event2 - 等待所有事件（AllOf）
• event1 | event2 - 等待任何事件（AnyOf）

资源

SimPy为不同场景提供多种资源类型。有关详细信息，请参阅references/resources.md。

资源类型摘要

资源类型	用例
Resource	有限容量（服务器、机器）
PriorityResource	基于优先级的排队
PreemptiveResource	高优先级可中断低优先级
Container	散装材料（燃料、水）
Store	Python对象存储（FIFO）
FilterStore	选择性物品检索
PriorityStore	优先级排序物品

快速参考

import simpy

env = simpy.Environment()

# 基本资源（例如，服务器）
resource = simpy.Resource(env, capacity=2)

# 优先级资源
priority_resource = simpy.PriorityResource(env, capacity=1)

# 容器（例如，燃料箱）
fuel_tank = simpy.Container(env, capacity=100, init=50)

# 存储（例如，仓库）
warehouse = simpy.Store(env, capacity=10)

常见模拟模式

模式1: 客户-服务器队列

import simpy
import random

def customer(env, name, server):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        print(f'{name} waited {wait:.2f}, served at {env.now}')
        yield env.timeout(random.uniform(2, 4))

def customer_generator(env, server):
    i = 0
    while True:
        yield env.timeout(random.uniform(1, 3))
        i += 1
        env.process(customer(env, f'Customer {i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(customer_generator(env, server))
env.run(until=20)

模式2: 生产者-消费者

import simpy

def producer(env, store):
    item_id = 0
    while True:
        yield env.timeout(2)
        item = f'Item {item_id}'
        yield store.put(item)
        print(f'Produced {item} at {env.now}')
        item_id += 1

def consumer(env, store):
    while True:
        item = yield store.get()
        print(f'Consumed {item} at {env.now}')
        yield env.timeout(3)

env = simpy.Environment()
store = simpy.Store(env, capacity=10)
env.process(producer(env, store))
env.process(consumer(env, store))
env.run(until=20)

模式3: 并行任务执行

import simpy

def task(env, name, duration):
    print(f'{name} starting at {env.now}')
    yield env.timeout(duration)
    print(f'{name} done at {env.now}')
    return f'{name} result'

def coordinator(env):
    # 并行启动任务
    task1 = env.process(task(env, 'Task 1', 5))
    task2 = env.process(task(env, 'Task 2', 3))
    task3 = env.process(task(env, 'Task 3', 4))

    # 等待所有完成
    results = yield task1 & task2 & task3
    print(f'All done at {env.now}')

env = simpy.Environment()
env.process(coordinator(env))
env.run()

工作流程指南

步骤1: 定义系统

识别：

• 实体: 什么通过系统移动？（客户、零件、数据包）
• 资源: 约束是什么？（服务器、机器、带宽）
• 过程: 活动是什么？（到达、服务、离开）
• 指标: 测量什么？（等待时间、利用率、吞吐量）

步骤2: 实现过程函数

为每种过程类型创建生成器函数：

def entity_process(env, name, resources, parameters):
    # 到达逻辑
    arrival_time = env.now

    # 请求资源
    with resource.request() as req:
        yield req

        # 服务逻辑
        service_time = calculate_service_time(parameters)
        yield env.timeout(service_time)

    # 离开逻辑
    collect_statistics(env.now - arrival_time)

步骤3: 设置监控

使用监控工具收集数据。有关技术，请参阅references/monitoring.md。

from scripts.resource_monitor import ResourceMonitor

# 创建并监控资源
resource = simpy.Resource(env, capacity=2)
monitor = ResourceMonitor(env, resource, "Server")

# 模拟后
monitor.report()

步骤4: 运行和分析

# 运行模拟
env.run(until=simulation_time)

# 生成报告
monitor.report()
stats.report()

# 导出数据用于进一步分析
monitor.export_csv('results.csv')

高级功能

过程交互

过程可以通过事件、过程产生和中断进行交互。有关模式，请参阅references/process-interaction.md。

关键机制:

• 事件信号: 用于协调的共享事件
• 过程产生: 等待其他过程完成
• 中断: 强制恢复过程以进行抢占

实时模拟

与挂钟时间同步模拟，用于硬件在环或交互应用。请参阅references/real-time.md。

import simpy.rt

env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1:1时间映射
# factor=0.5表示1模拟单位 = 0.5秒（快2倍）

全面监控

监控过程、资源和事件。有关技术，请参阅references/monitoring.md，包括：

• 状态变量跟踪
• 资源猴子补丁
• 事件追踪
• 统计收集

脚本和模板

basic_simulation_template.py

构建队列模拟的完整模板，包括：

• 可配置参数
• 统计收集
• 客户生成
• 资源使用
• 报告生成

用法:

from scripts.basic_simulation_template import SimulationConfig, run_simulation

config = SimulationConfig()
config.num_resources = 2
config.sim_time = 100
stats = run_simulation(config)
stats.report()

resource_monitor.py

可重用监控工具：

• ResourceMonitor - 跟踪单个资源
• MultiResourceMonitor - 监控多个资源
• ContainerMonitor - 跟踪容器水平
• 自动统计计算
• CSV导出功能

用法:

from scripts.resource_monitor import ResourceMonitor

monitor = ResourceMonitor(env, resource, "My Resource")
# ... 运行模拟 ...
monitor.report()
monitor.export_csv('data.csv')

参考文档

特定主题的详细指南：

• references/resources.md - 所有资源类型及示例
• references/events.md - 事件系统和模式
• references/process-interaction.md - 过程同步
• references/monitoring.md - 数据收集技术
• references/real-time.md - 实时模拟设置

最佳实践

1. 生成器函数: 始终在过程函数中使用yield
2. 资源上下文管理器: 使用with resource.request() as req: 进行自动清理
3. 可再现性: 设置random.seed() 以获得一致结果
4. 监控: 在整个模拟过程中收集数据，而不仅仅是最后
5. 验证: 将简单案例与解析解比较
6. 文档化: 注释过程逻辑和参数选择
7. 模块化设计: 分离过程逻辑、统计和配置

常见陷阱

1. 忘记yield: 过程必须产生事件以暂停
2. 事件重用: 事件只能触发一次
3. 资源泄漏: 使用上下文管理器或确保释放
4. 阻塞操作: 避免在过程中使用Python阻塞调用
5. 时间单位: 保持时间单位解释一致
6. 死锁: 确保至少一个过程可以取得进展

示例用例

• 制造: 机器调度、生产线、库存管理
• 医疗保健: 急诊室模拟、患者流、人员分配
• 电信: 网络流量、数据包路由、带宽分配
• 交通: 交通流、物流、车辆路由
• 服务操作: 呼叫中心、零售结账、预约调度
• 计算机系统: CPU调度、内存管理、I/O操作