计划执行与检查节点
executing-plans
按既定实施计划完成工作,保持过程检查和反馈。
simpy【效率与协作】A process-based discrete-event simulation framework. Use for modeling queuing systems, supply chains, manufacturing processes, network simulation, project management, and any system where events occur at specific points in time. Load when working with discrete event simulation, process modeling, resource allocation, virtual time, simpy.Environment, s…
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
SimPy allows you to model real-world processes as Python generators. Components in SimPy (like customers, cars, or data packets) are "processes" that interact with each other and with limited "resources" (like servers, parking spots, or bandwidth).
Official docs: https://simpy.readthedocs.io/
GitHub: https://github.com/simpy/simpy
Search patterns: simpy.Environment, simpy.Resource, env.process, yield env.timeout
SimPy uses an internal clock starting at 0. Time moves forward only when an event is processed. Between events, "nothing" happens, so simulating 100 years takes seconds if only a few events occur.
Processes are standard Python functions using yield. When a process yields an event, SimPy suspends it until that event occurs.
Three types of shared objects:
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) - This is the only way to advance time for a process. Never use time.sleep().with resource.request() as req: - This ensures a resource is automatically released even if the process is interrupted or fails.random.seed(42) for reproducible simulations.env object - Every process and resource needs a reference to the environment to track time.env.run(until=...) - Always define a stopping condition, otherwise a while True simulation will run forever.env.now is your only clock.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'
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()
# 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
# 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
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 ...
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()
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())
In systems where some events are more important (e.g., critical patients), use PriorityResource.
res = simpy.PriorityResource(env, capacity=1)
with res.request(priority=0) as req: # Lower number = higher priority
yield req
Avoid printing to the console in high-speed simulations. Instead, use a data collection list and process it with NumPy/Pandas after the env.run() finishes.
If your code reaches a logic branch where no yield occurs in a while True loop, SimPy freezes.
# ❌ 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
Never mix objects (Resources, Processes) from different environments. Every component must belong to the same simpy.Environment() instance.
When interrupting a process, the interrupt happens at the current virtual time. Be careful with what the interrupted process was yielding (it will receive a simpy.Interrupt exception).
with resource.request() as req: ensures proper cleanuprandom.seed(42) at the start of simulationsenv.run(until=...) with a specific time limittime.sleep() or real-world time with SimPySimPy turns Python's generator syntax into a powerful engine for modeling time and scarcity. It allows scientists and engineers to experiment with "What if?" scenarios for complex systems without the cost or risk of physical trials.