mcpField
S
simpy

simpy

github.com/tondevrel数据与数据库2452 次安装39 周前更新
代码仓库

关于 simpy

SimPy 是用于离散事件模拟的 Python 技能，适合研究排队、物流、制造和网络等按事件推进的系统。它将流程表示为生成器，并用虚拟时间及共享资源模型描述系统运行。

主要功能

  1. 01生成器定义模拟流程
  2. 02事件驱动的虚拟时钟
  3. 03Resource、Container 与 Store
  4. 04事件同步与中断处理
  5. 05优先级资源管理

适用场景

  1. 01模拟银行柜台或急诊排队
  2. 02分析供应链与物流流程
  3. 03评估装配线和设备维修
  4. 04模拟网络数据包与服务器负载

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/tondevrel/scientific-agent-skills --skill simpy

安装后在 Claude Code 中输入/simpy确认已加载。

详细介绍

SimPy - 离散事件仿真

SimPy 可让你将现实世界的流程建模为 Python 生成器。SimPy 中的组件（如客户、汽车或数据包）是相互交互并与有限“资源”（如服务人员、停车位或带宽）交互的“进程”。

适用场景

  • 对排队系统建模（银行柜员、呼叫中心、医院急诊室）
  • 仿真供应链和物流（仓库、运输网络）
  • 分析制造流程（装配线、机器维护）
  • 网络仿真（数据包路由、服务器负载均衡）
  • 项目管理（任务依赖关系、资源分配）
  • 任何事件在特定时间点发生，而非连续发生的系统

参考文档

官方文档：https://simpy.readthedocs.io/
GitHub：https://github.com/simpy/simpy
搜索模式：simpy.Environment、simpy.Resource、env.process、yield env.timeout

核心原则

虚拟时间

SimPy 使用从 0 开始的内部时钟。只有处理事件时，时间才会向前推进。事件之间什么也不会发生，因此如果只发生少量事件，模拟 100 年也只需几秒钟。

作为生成器的进程

进程是使用 yield 的标准 Python 函数。进程让出一个事件时，SimPy 会将其挂起，直到该事件发生。

资源

三种共享对象：

  • Resource：数量有限的服务槽位（例如柜台）
  • Container：用于大宗物质（例如储气罐、RAM）
  • Store：用于独立对象（例如消息缓冲区）

快速参考

安装
pip install simpy
标准导入
import simpy
import random
基本模式 - 简单进程
import simpy

def clock(env, name, tick):
    while True:
        print(f"{name} at {env.now}")
        # 'yield' tells SimPy to wait for this event
        yield env.timeout(tick)

# 1. Create Environment
env = simpy.Environment()

# 2. Add Process
env.process(clock(env, 'Fast', 0.5))
env.process(clock(env, 'Slow', 1.0))

# 3. Run for 2 units of virtual time
env.run(until=2.1)

重要规则

✅ 应该做
  • 使用 yield env.timeout(n) - 这是推进进程时间的唯一方式。切勿使用 time.sleep()。
  • 使用 with resource.request() as req: - 这样即使进程被中断或失败，也能自动释放资源。
  • 设置随机种子 - 始终使用 random.seed(42)，确保模拟结果可复现。
  • 传递 env 对象 - 每个进程和资源都需要引用环境，以跟踪时间。
  • 使用 env.run(until=...) - 始终定义停止条件，否则 while True 模拟将永远运行。
  • 将数据收集到列表中 - 将时间戳和事件详情存储到外部列表中，以便在模拟后进行分析（例如使用 Pandas）。
❌ 不要做
  • 使用现实世界时间 - simpy 用于虚拟时间。env.now 是你唯一的时钟。
  • 直接修改环境 - 只能通过进程和事件与环境交互。
  • 忘记让出事件 - 如果循环中没有让出事件，就会在虚拟时间的同一时刻陷入无限循环，导致模拟挂起。
  • 阻塞事件循环 - 进程中应尽量避免繁重的 CPU 计算；CPU 运算期间，整个模拟的虚拟时钟会“停止”。

反模式（绝对不要）

import simpy
import time

# ❌ BAD: Using real sleep
def bad_process(env):
    time.sleep(1) # This stops the REAL clock, but virtual time doesn't move!
    yield env.timeout(1)

# ✅ GOOD: Use virtual timeout
def good_process(env):
    yield env.timeout(1) # Moves virtual clock forward instantly

# ❌ BAD: Infinite loop without yielding
# def hang_forever(env):
#     while True:
#         x = 1 + 1 # No yield = simulation clock stuck at current env.now

# ❌ BAD: Manual resource release (Dangerous)
def risky_request(env, res):
    req = res.request()
    yield req
    # If something fails here, the resource is never released
    res.release(req)

# ✅ GOOD: Using context manager
def safe_request(env, res):
    with res.request() as req:
        yield req
        # Resource automatically released when exiting 'with'

资源与队列（simpy.Resource）

管理共享容量
def car(env, name, bcs, driving_time, charge_time):
    # Simulate driving to the station
    yield env.timeout(driving_time)

    print(f'{name} arriving at {env.now}')
    # Request one of the charging spots
    with bcs.request() as req:
        yield req
        print(f'{name} starting to charge at {env.now}')
        yield env.timeout(charge_time)
        print(f'{name} leaving at {env.now}')

env = simpy.Environment()
# Battery Charging Station with 2 spots
bcs = simpy.Resource(env, capacity=2)

for i in range(4):
    env.process(car(env, f'Car {i}', bcs, i*2, 5))

env.run()

高级资源：Container 和 Store

Container（液体/大宗物质）
# Modeling a gas station tank
gas_tank = simpy.Container(env, capacity=1000, init=500)

def refuel(env, gas_tank):
    yield gas_tank.put(400) # Add fuel

def car(env, gas_tank):
    yield gas_tank.get(40) # Consume fuel
Store（独立对象）
# Modeling a producer-consumer buffer
buffer = simpy.Store(env, capacity=10)

def producer(env, buffer):
    yield buffer.put(f'Widget at {env.now}')

def consumer(env, buffer):
    item = yield buffer.get()
    print(f'Consumed {item}')

事件同步

等待多个事件
# Wait for BOTH events
yield env.timeout(10) & other_process_event

# Wait for EITHER event
yield env.timeout(10) | other_process_event

实用工作流

1. 医院急诊室模型
def patient(env, name, nurse, doctor):
    arrival_time = env.now
    
    # 1. Wait for Nurse (Triage)
    with nurse.request() as req:
        yield req
        yield env.timeout(random.expovariate(1/5)) # 5 mins avg
        
    # 2. Wait for Doctor (Treatment)
    with doctor.request() as req:
        yield req
        yield env.timeout(random.normalvariate(20, 5)) # 20 mins avg
        
    wait_times.append(env.now - arrival_time)

wait_times = []
env = simpy.Environment()
nurse = simpy.Resource(env, capacity=2)
doctor = simpy.Resource(env, capacity=1)
# ... launch processes ...
2. 机器故障与维修
def machine(env, repairman):
    while True:
        try:
            # Working phase
            working_time = 100
            yield env.timeout(working_time)
        except simpy.Interrupt:
            # Interrupted by failure
            with repairman.request() as req:
                yield req
                yield env.timeout(10) # Repair time

def break_machine(env, machine_proc):
    while True:
        yield env.timeout(random.uniform(50, 200))
        machine_proc.interrupt()
3. 服务器请求处理（数据收集）
import pandas as pd

log = []

def request(env, server):
    start_wait = env.now
    with server.request() as req:
        yield req
        wait = env.now - start_wait
        service_time = random.expovariate(1.0)
        yield env.timeout(service_time)
        
        log.append({
            'arrival': start_wait,
            'wait': wait,
            'service': service_time,
            'finish': env.now
        })

# After simulation:
# df = pd.DataFrame(log)
# print(df['wait'].mean())

性能优化

使用 simpy.PriorityResource

在某些事件更重要的系统中（例如危重患者），使用 PriorityResource。

res = simpy.PriorityResource(env, capacity=1)
with res.request(priority=0) as req: # Lower number = higher priority
    yield req
监控大型模拟

高速模拟中应避免向控制台打印内容。应改用数据收集列表，并在 env.run() 完成后使用 NumPy/Pandas 处理数据。

常见陷阱与解决方案

“卡住的进程”

如果代码进入 while True 循环中没有 yield 的逻辑分支，SimPy 就会冻结。

# ❌ Problem:
def logic_error(env):
    while True:
        if condition:
            yield env.timeout(1)
        # Else? If condition is false, we loop forever at same time point!

# ✅ Solution:
def logic_fixed(env):
    while True:
        if condition:
            yield env.timeout(1)
        else:
            yield env.timeout(0.1) # Or some default wait
多个环境

切勿混用来自不同环境的对象（资源、进程）。每个组件都必须属于同一个 simpy.Environment() 实例。

中断时机

中断进程时，中断会在当前虚拟时间发生。请注意被中断的进程正在等待什么（它会收到 simpy.Interrupt 异常）。

最佳实践

  1. 始终为资源使用上下文管理器 - with resource.request() as req: 可确保正确清理
  2. 设置随机种子以确保可复现性 - 在模拟开始时使用 random.seed(42)
  3. 在外部收集数据 - 将事件数据存储在列表/字典中，而不只是打印出来
  4. 定义明确的停止条件 - 始终使用 env.run(until=...) 并指定时间限制
  5. 仅使用虚拟时间 - 切勿将 time.sleep() 或现实世界的时间与 SimPy 混用
  6. 先用小规模模拟进行测试 - 在运行大规模模拟前验证逻辑
  7. 记录进程文档 - 为进程和资源清晰命名，以便调试
  8. 正确处理中断 - 对可能被中断的进程使用 try/except 代码块

SimPy 将 Python 的生成器语法转化为强大的时间与稀缺性建模引擎。它让科学家和工程师能够针对复杂系统开展“假设”场景实验，而无需承担实体试验的成本或风险。

评论

支持 Markdown