by davila7
没有模拟很难分析具有队列和共享资源的复杂系统。SimPy 提供了一个基于 Python 的框架,用于对流程、资源和时间事件进行建模,以便在实施前预测系统行为。
1. 打开 Claude 聊天界面
2. 点击下方 "📋 复制" 按钮
3. 粘贴到 Claude 聊天框中并发送
4. 输入 "使用 simpy 技能" 开始使用
=== simpy 技能 === 作者: davila7 描述: 没有模拟很难分析具有队列和共享资源的复杂系统。SimPy 提供了一个基于 Python 的框架,用于对流程、资源和时间事件进行建模,以便在实施前预测系统行为。 使用方法: 1. 调用技能: "使用 simpy 技能" 2. 提供相关信息: 根据技能要求提供必要参数 3. 查看结果: 技能会返回处理结果 示例: "使用 simpy 技能,帮我分析一下这段代码"
这种方法适用于所有 Claude 用户,不需要安装额外工具。
research
safe
SimPy is a process-based discrete-event simulation framework based on standard Python. Use SimPy to model systems where entities (customers, vehicles, packets, etc.) interact with each other and compete for shared resources (servers, machines, bandwidth, etc.) over time.
Core capabilities:
Use the SimPy skill when:
Not suitable for:
import simpy
def process(env, name):
"""A simple process that waits and prints."""
print(f'{name} starting at {env.now}')
yield env.timeout(5)
print(f'{name} finishing at {env.now}')
# Create environment
env = simpy.Environment()
# Start processes
env.process(process(env, 'Process 1'))
env.process(process(env, 'Process 2'))
# Run simulation
env.run(until=10)
import simpy
def customer(env, name, resource):
"""Customer requests resource, uses it, then releases."""
with resource.request() as req:
yield req # Wait for resource
print(f'{name} got resource at {env.now}')
yield env.timeout(3) # Use resource
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()
The simulation environment manages time and schedules events.
import simpy
# Standard environment (runs as fast as possible)
env = simpy.Environment(initial_time=0)
# Real-time environment (synchronized with wall-clock)
import simpy.rt
env_rt = simpy.rt.RealtimeEnvironment(factor=1.0)
# Run simulation
env.run(until=100) # Run until time 100
env.run() # Run until no events remain
Processes are defined using Python generator functions (functions with yield statements).
def my_process(env, param1, param2):
"""Process that yields events to pause execution."""
print(f'Starting at {env.now}')
# Wait for time to pass
yield env.timeout(5)
print(f'Resumed at {env.now}')
# Wait for another event
yield env.timeout(3)
print(f'Done at {env.now}')
return 'result'
# Start the process
env.process(my_process(env, 'value1', 'value2'))
Events are the fundamental mechanism for process synchronization. Processes yield events and resume when those events are triggered.
Common event types:
env.timeout(delay) - Wait for time to passresource.request() - Request a resourceenv.event() - Create a custom eventenv.process(func()) - Process as an eventevent1 & event2 - Wait for all events (AllOf)event1 | event2 - Wait for any event (AnyOf)SimPy provides several resource types for different scenarios. For comprehensive details, see references/resources.md.
| Resource Type | Use Case |
|---|---|
| Resource | Limited capacity (servers, machines) |
| PriorityResource | Priority-based queuing |
| PreemptiveResource | High-priority can interrupt low-priority |
| Container | Bulk materials (fuel, water) |
| Store | Python object storage (FIFO) |
| FilterStore | Selective item retrieval |
| PriorityStore | Priority-ordered items |
import simpy
env = simpy.Environment()
# Basic resource (e.g., servers)
resource = simpy.Resource(env, capacity=2)
# Priority resource
priority_resource = simpy.PriorityResource(env, capacity=1)
# Container (e.g., fuel tank)
fuel_tank = simpy.Container(env, capacity=100, init=50)
# Store (e.g., warehouse)
warehouse = simpy.Store(env, capacity=10)
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)
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)
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):
# Start tasks in parallel
task1 = env.process(task(env, 'Task 1', 5))
task2 = env.process(task(env, 'Task 2', 3))
task3 = env.process(task(env, 'Task 3', 4))
# Wait for all to complete
results = yield task1 & task2 & task3
print(f'All done at {env.now}')
env = simpy.Environment()
env.process(coordinator(env))
env.run()
Identify:
Create generator functions for each process type:
def entity_process(env, name, resources, parameters):
# Arrival logic
arrival_time = env.now
# Request resources
with resource.request() as req:
yield req
# Service logic
service_time = calculate_service_time(parameters)
yield env.timeout(service_time)
# Departure logic
collect_statistics(env.now - arrival_time)
Use monitoring utilities to collect data. See references/monitoring.md for comprehensive techniques.
from scripts.resource_monitor import ResourceMonitor
# Create and monitor resource
resource = simpy.Resource(env, capacity=2)
monitor = ResourceMonitor(env, resource, "Server")
# After simulation
monitor.report()
# Run simulation
env.run(until=simulation_time)
# Generate reports
monitor.report()
stats.report()
# Export data for further analysis
monitor.export_csv('results.csv')
Processes can interact through events, process yields, and interrupts. See references/process-interaction.md for detailed patterns.
Key mechanisms:
Synchronize simulation with wall-clock time for hardware-in-the-loop or interactive applications. See references/real-time.md.
import simpy.rt
env = simpy.rt.RealtimeEnvironment(factor=1.0) # 1:1 time mapping
# factor=0.5 means 1 sim unit = 0.5 seconds (2x faster)
Monitor processes, resources, and events. See references/monitoring.md for techniques including:
Complete template for building queue simulations with:
Usage:
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()
Reusable monitoring utilities:
ResourceMonitor - Track single resourceMultiResourceMonitor - Monitor multiple resourcesContainerMonitor - Track container levelsUsage:
from scripts.resource_monitor import ResourceMonitor
monitor = ResourceMonitor(env, resource, "My Resource")
# ... run simulation ...
monitor.report()
monitor.export_csv('data.csv')
Detailed guides for specific topics:
references/resources.md - All resource types with examplesreferences/events.md - Event system and patternsreferences/process-interaction.md - Process synchronizationreferences/monitoring.md - Data collection techniquesreferences/real-time.md - Real-time simulation setupyield in process functionswith resource.request() as req: for automatic cleanuprandom.seed() for consistent resultsView Count
0
Download Count
0
Favorite Count
0
Quality Score
77