mcpField
S
simpy

simpy

github.com/kalarislabs其他7343 次安装4 天前更新
代码仓库

关于 simpy

simpy 是一项用于 Python 进程式离散事件仿真的技能,适合构建和评估事件驱动模型。内容涵盖 SimPy 的事件与资源语义、有限运行、监测及仿真研究方法。

主要功能

  1. 01事件驱动的进程建模
  2. 02共享资源与队列原语
  3. 03有限仿真执行
  4. 04监测与事件追踪
  5. 05重复运行与验证指南
  6. 06安全且有界的仿真 CLI

适用场景

  1. 01模拟服务队列和生产线
  2. 02模拟物流与库存系统
  3. 03测试资源竞争和路由
  4. 04运行有界队列场景与重复试验

安装

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

$npx skills add https://github.com/kalarislabs/research-agent-skills --skill simpy

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

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

references/
cli-guide.md
events.md
monitoring.md
process-interaction.md
real-time.md
resources.md
simulation-methodology.md
sources.md
scripts/
_common.py
basic_simulation_template.py
bounded_queue_scenario.py
event_trace_summary.py
replication_runner.py
resource_monitor.py
validate_simulation_config.py

详细介绍

SimPy

适用范围

将此技能用于基于流程的离散事件模型,其中活跃实体会让出事件并竞争资源:队列、生产系统、物流、网络、服务运营、库存及其他事件驱动系统。

SimPy 提供事件调度器和建模原语。它不会替你选择科学上有效的概念模型、输入分布、预热期、运行时长、重复次数、估计目标或因果解释。应将这些视为仿真研究方法,而不是 SimPy API 的行为。

当前版本与安装

已于 2026-07-23 验证:

  • 最新稳定版本:SimPy 4.1.2,于 2026-05-24 在 PyPI 发布;源码标签 4.1.2 指向提交 f4381649。
  • 软件包元数据要求 Python >=3.8,并标明支持 CPython 3.8-3.14 和 PyPy。SimPy 没有运行时依赖项。
  • 4.1.2 增加了对 Python 3.13/3.14 的支持,并修复了现代解释器相关的测试问题。
  • 上游项目和此技能均采用 MIT 许可证。

创建可复现的环境:

uv venv --python 3.13
source .venv/bin/activate
uv pip install "simpy==4.1.2"
python -c "import importlib.metadata; print(importlib.metadata.version('simpy'))"

不要默默替换为 latest 文档构建版本:它可能描述尚未发布的开发版本。请使用 references/sources.md 中的 4.1.2 版本链接。

模型工作流

  1. 定义目的和估计目标。 说明决策/问题、系统边界、实体、资源、状态、输出、时间单位,以及终止事件或稳态目标。
  2. 先编写概念模型。 记录假设、分布、路由、优先级、初始条件和省略的机制。
  3. 实现生成器。 SimPy 流程是一个会让出事件的 Python 生成器。使用 env.process(...) 注册生成器对象。
  4. 限定执行范围。 为每次生产运行明确设置时间、实体、事件和重复次数上限。切勿对包含无限流程的模型调用 env.run()。
  5. 分离随机流。 对逻辑上不同的随机来源使用独立的本地 RNG 实例;保留种子清单。
  6. 有意设置监测。 在目标状态转换之后观察状态,在时间范围终点结算时间加权区间,并测试监测是否改变事件顺序。
  7. 验证与确认。 测试确定性边界情形、守恒恒等式、事件轨迹、队列规则和解析基准;针对既定目的,与系统证据或专家证据进行比较。
  8. 运行独立重复试验。 根据每次重复试验的估计值计算区间,而不是使用单次运行中相互关联的实体。
  9. 报告局限性。 包括初始化、未完成的实体、运行时长、种子/随机流、精度、敏感性和验证证据。切勿将仿真中的关联转化为因果主张。

提出推断性主张前,请阅读 references/simulation-methodology.md。

最小有界模型

import random
import simpy

HORIZON = 480.0
arrival_rng = random.Random(101)
service_rng = random.Random(202)
env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
completed = []

def customer(arrival):
    with server.request() as request:
        yield request
        wait = env.now - arrival
        yield env.timeout(service_rng.expovariate(1 / 6.0))
    completed.append((env.now, wait))

def arrivals():
    for _ in range(10_000):  # Entity cap.
        delay = arrival_rng.expovariate(1 / 4.0)
        if env.now + delay >= HORIZON:
            return
        yield env.timeout(delay)
        env.process(customer(env.now))

env.process(arrivals())
env.run(until=HORIZON)

数值时间范围采用左闭右开区间:恰好安排在 480.0 的常规事件不会被处理。应报告未完成的实体,而不是默默将其视为已完成的观测。

核心语义

环境与确定性排序

Environment 是单线程的。队列按仿真时间、事件优先级排序,然后按严格递增的事件 ID 排序。因此,同一时间、同一优先级的事件会按调度顺序先进先出地处理。模型进程可以表示并发,但回调会按顺序确定性地执行。

  • env.now:无单位的仿真时钟;请选择并记录一个时间单位。
  • env.peek():下一个事件的时间,或无穷大。
  • env.step():处理一个事件;队列为空时会引发 EmptySchedule。
  • env.active_process:当前正在执行的进程,否则为 None。
  • env.run():清空队列;存在循环或无限进程时使用不安全。

env.run(until=number) 和 env.run(until=event) 在边界处并不等价:

  • 数值会调度一个紧急停止事件,并排除恰好发生在该时间的普通事件。
  • Event 条件会在该事件的停止回调触发时返回其值。其他同一时间事件的顺序取决于优先级和调度顺序。
  • 在 4.1.2 中,Environment.step() 会通过重新调度目标事件,保留 StopSimulation 之后剩余的回调。因此,env.run(until=target) 返回后,即使目标事件的值已返回,target.processed 仍可能是 False,直到再执行一次 step()/run()。不要将 processed 作为运行后判断完成状态的唯一依据。

请参阅 references/events.md 和 references/monitoring.md。

Event、Timeout、Process 和 Condition
  • Event 只会经历一次以下状态变化:未触发 -> 已触发/已调度 -> 已处理。succeed(value) 或 fail(exception) 会触发它一次。
  • Timeout 在创建时触发,并安排在 now + delay 时发生;之后无法再次手动使其成功。
  • env.process(generator) 会创建一个 Process;生成器恢复执行时会收到所产出的事件值。生成器返回时,Process 会以该返回值成功结束。未捕获的异常会导致其失败。
  • AnyOf / a | b 和 AllOf / a & b 会产生一个 ConditionValue:这是一个有序、类似字典的映射,其键为事件对象,值为对应事件的值。请使用原始事件对象检查成员关系;不要假定结果是标量。
  • AnyOf 不会取消未获胜的事件。放弃资源请求时,请显式取消尚未处理的请求;普通超时仍会保持已调度状态。
中断

process.interrupt(cause) 会调度一个紧急中断,将 simpy.Interrupt 注入目标生成器。在可能被中断的已产出工作代码周围捕获该异常,检查 interrupt.cause,更新剩余工作,然后选择继续执行、重新产出原事件或终止。

中断进程会从其当前目标事件中移除恢复回调,但不会取消该目标事件。进程不能中断自身或已终止的进程。请参阅 references/process-interaction.md。

共享资源

类型语义
Resource类似信号量的 FIFO 使用名额
PriorityResource按数值优先级从低到高对排队请求排序
PreemptiveResource优先级队列,并可选择抢占当前使用者
Container同质数值级别;put/get 会等待容量/物料
StoreFIFO Python 对象
FilterStore首个符合请求谓词的可用项
PriorityStore按优先级顺序返回可比较的项

使用请求上下文管理器:

def job(env, resource):
    with resource.request() as request:
        yield request
        yield env.timeout(3)

退出时,它会释放已获取的请求,或取消仍在等待的请求,包括异常展开期间。对于手动保留的待处理 put/get/请求,如果中断或超时导致进程放弃该请求,请调用 cancel()。

PreemptiveResource.request(priority=..., preempt=True) 使用较小的数字表示较高的优先级。被抢占的进程会收到一个 Interrupt,其原因是一个 Preempted 对象:cause.by 是发起抢占的 Process,cause.usage_since 是开始使用的时间,cause.resource 是该资源。排队优先级优先于 preempt 标志;混用抢占与非抢占请求时,需要进行明确的测试。

请阅读 references/resources.md,了解被阻塞的操作、队列规则和示例。

监控与步进

优先在状态转换时进行明确的领域观测。对于通用资源监控,可通过包装器或子类检查 count、queue、level、items、put_queue 和 get_queue。对于事件跟踪,schedule() 和 step() 是核心钩子。

队列测量对时序很敏感:

  • 请求方法调用前的状态、调用后的状态、授予回调和释放回调,在同一仿真时间戳下都可能不同。
  • 样本均值对事件观测加权,而非对时间加权。计算左连续状态路径下的曲线面积,并除以经过的时间。
  • 添加初始和最终样本;在分析时域处结束最后一个时间区间。
  • env._queue、资源 _env 和猴子补丁都是实现细节。固定 SimPy 版本、隔离监控代码,并在升级后进行回归测试。
  • 跟踪每个事件会增加运行时间和内存占用;请限制跟踪记录数量。

请使用 scripts/resource_monitor.py 和 references/monitoring.md。

实时执行

simpy.rt.RealtimeEnvironment(initial_time=0, factor=1.0, strict=True) 将一个仿真时间单位映射为 factor 秒的实际时间。在严格模式下,如果计算落后,step()/run() 会引发 RuntimeError。strict=False 会容忍延迟,但不会恢复计时精度。使用 Environment 开发逻辑,然后在单独的计时测试中采用适用于平台且较宽松的容差。请参阅 references/real-time.md。

附带的安全 CLI

所有 CLI 都使用固定的内置队列模型,或汇总本地构件。它们会拒绝未知的 JSON 键、URL、符号链接、非有限数、超大输入,以及无界的时间/事件/实体/重复次数。它们绝不会评估配置文本、执行用户 Python、导入插件或调用网络服务。

# Inspect all options.
python skills/simpy/scripts/bounded_queue_scenario.py --help
python skills/simpy/scripts/replication_runner.py --help
python skills/simpy/scripts/event_trace_summary.py --help
python skills/simpy/scripts/validate_simulation_config.py --help

# Deterministic built-in scenario.
python skills/simpy/scripts/bounded_queue_scenario.py

# Independent replications with replication-level Student-t intervals.
python skills/simpy/scripts/replication_runner.py

# Validate only; no simulation runs.
python skills/simpy/scripts/validate_simulation_config.py config.json

重复运行器会拒绝仅有一次重复的区间。其区间量化的是配置模型下的 Monte Carlo 不确定性;它们既不能验证模型,也不能识别因果效应。请参阅 references/cli-guide.md。

测试

使用确定性单元测试覆盖排序、边界时间、条件、中断、所有资源调度策略、守恒、事件/实体限制、种子可复现性和监控的非干扰性。仅添加用于宽泛分布检查的随机测试,并固定种子;避免脆弱的精确样本估计。

请在完全匹配的固定环境中运行此技能的测试套件,且不生成字节码构件:

PYTHONDONTWRITEBYTECODE=1 uv run --isolated --no-project \
  --python 3.13 --with "simpy==4.1.2" \
  python -m unittest discover -s tests/simpy -v

参考资料

  • references/events.md — 调度器、生命周期、运行边界、条件
  • references/process-interaction.md — 生成器、共享事件、中断
  • references/resources.md — 所有 Resource、Container 和 Store 变体
  • references/monitoring.md — 时间加权、队列计时、追踪、步进
  • references/real-time.md — 因子、严格模式、漂移、计时测试
  • references/simulation-methodology.md — 重复运行、预热、验证、CI
  • references/cli-guide.md — 架构、边界、输出和安全的 CLI 示例
  • references/sources.md — 标注日期的官方来源和一手方法来源

评论

支持 Markdown