简单的资源容量管理
Simpy resource capacity management
我正在为一个非常长的多次重入制造过程开发一个离散偶数模拟器（零件按特定顺序经过多个工具，并且经常多次返回到同一个工具）。该程序的整体流程运行良好，并在各种构建计划场景下给出了预期的队列深度、拥塞点等。我还没有找到解决方法的挑战是容量 >1 的工具可能 运行 ，一次多个部分，但它必须同时启动它们（即它们可能 运行 在房间或浴缸中，例如当一个部分是 运行 时不能 opened/accessed 添加另一个）。
因此，我正在寻找一种方法来实现一个工具，比如 capacity=4，这样在 运行 开始时，如果队列中有 4 个或更多项目，它'我将加载其中的 4 个和 运行 它们，但是如果在时间为零时只有一个部分，则该部分 运行s 以及进入队列的任何东西，而该部分是 运行 ning要等到运行结束
我的代码太长太复杂，无法说明问题，但是著名的简单“加油站”示例（代码如下）很好地描述了问题。这段代码中表达了我试图摆脱的行为。也就是说，加油站的容量为 2，一辆车出现并占据了其中一个槽位，然后一段时间后另一辆车到达并占据了剩余的槽位。这对加油站来说很棒，但我试图阻止以后的用户在 运行 启动后获得访问权限。
我可以设想为该工具提供一个 属性，如 self.status，并在使用该工具时将状态设置为 'busy' 或类似的东西，或者可能使用 self.res.users 以某种方式做到这一点，但我想知道是否有一些更简单的方式让我的工具以所需的方式运行。
谢谢！
import itertools
import random
import simpy
RANDOM_SEED = 42
GAS_STATION_SIZE = 200 # liters
THRESHOLD = 10 # Threshold for calling the tank truck (in %)
FUEL_TANK_SIZE = 50 # liters
FUEL_TANK_LEVEL = [5, 25] # Min/max levels of fuel tanks (in liters)
REFUELING_SPEED = 2 # liters / second
TANK_TRUCK_TIME = 300 # Seconds it takes the tank truck to arrive
T_INTER = [30, 300] # Create a car every [min, max] seconds
SIM_TIME = 1000 # Simulation time in seconds
def car(name, env, gas_station, fuel_pump):
"""A car arrives at the gas station for refueling.
It requests one of the gas station's fuel pumps and tries to get the
desired amount of gas from it. If the stations reservoir is
depleted, the car has to wait for the tank truck to arrive.
"""
fuel_tank_level = random.randint(*FUEL_TANK_LEVEL)
print('%s arriving at gas station at %.1f' % (name, env.now))
with gas_station.request() as req:
start = env.now
# Request one of the gas pumps
yield req
# Get the required amount of fuel
liters_required = FUEL_TANK_SIZE - fuel_tank_level
yield fuel_pump.get(liters_required)
# The "actual" refueling process takes some time
yield env.timeout(liters_required / REFUELING_SPEED)
print('%s finished refueling in %.1f seconds.' % (name,
env.now - start))
def gas_station_control(env, fuel_pump):
"""Periodically check the level of the *fuel_pump* and call the tank
truck if the level falls below a threshold."""
while True:
if fuel_pump.level / fuel_pump.capacity * 100 < THRESHOLD:
# We need to call the tank truck now!
print('Calling tank truck at %d' % env.now)
# Wait for the tank truck to arrive and refuel the station
yield env.process(tank_truck(env, fuel_pump))
yield env.timeout(10) # Check every 10 seconds
def tank_truck(env, fuel_pump):
"""Arrives at the gas station after a certain delay and refuels it."""
yield env.timeout(TANK_TRUCK_TIME)
print('Tank truck arriving at time %d' % env.now)
amount = fuel_pump.capacity - fuel_pump.level
print('Tank truck refuelling %.1f liters.' % amount)
yield fuel_pump.put(amount)
def car_generator(env, gas_station, fuel_pump):
"""Generate new cars that arrive at the gas station."""
for i in itertools.count():
yield env.timeout(random.randint(*T_INTER))
env.process(car('Car %d' % i, env, gas_station, fuel_pump))
# Setup and start the simulation
print('Gas Station refuelling')
random.seed(RANDOM_SEED)
# Create environment and start processes
env = simpy.Environment()
gas_station = simpy.Resource(env, 2)
fuel_pump = simpy.Container(env, GAS_STATION_SIZE, init=GAS_STATION_SIZE)
env.process(gas_station_control(env, fuel_pump))
env.process(car_generator(env, gas_station, fuel_pump))
# Execute!
env.run(until=SIM_TIME)
所以在我看来，每辆车都抓住了一个泵，但你真正想要的是一个泵一次最多能抓住两辆车。所以你需要一个进行批处理的过程。这是一个快速而肮脏的批处理程序。我为输入队列使用一个存储，当我的每个 yield get() 获得第一个实体时，我检查队列是否有更多实体要处理，这个 yield 真正做的是导致我的进程在队列是空
"""
A simple example of a process that process entities in batches
Programmer: Michael R. Gibbs
"""
import simpy
import random
def batch_process(env, ent_q, next_q, max_cap):
"""
grabs up to max_cap of entites and processes them
"""
while True:
# use a yield to wait when queue is empty
ent_1 = yield ent_q.get()
# grabs up to capacity -1 more
p_batch = [ent_1] + ent_q.items[:(max_cap - 1)]
ent_q.items = ent_q.items[(max_cap - 1):]
print(f'{env.now:.2f} grabbed {len(p_batch)} entites leaving {len(ent_q.items)} left in queue')
# do the processing
yield env.timeout(10)
# send entitles to next stop
next_q.items = next_q.items + p_batch
def gen_ents(env, q):
"""
Generate the arrival of entities
arrival profile starts slow
has a peek
then nothing which drains the queue
and puts the process in a wait mode
another peek
"""
for _ in range(10):
yield env.timeout(random.uniform(1,6))
q.put(object())
for _ in range(15):
yield env.timeout(random.uniform(1,3))
q.put(object())
yield env.timeout(50)
for _ in range(15):
yield env.timeout(random.uniform(1,3))
q.put(object())
for _ in range(10):
yield env.timeout(random.uniform(1,7))
q.put(object())
# boot up model
random.seed(100)
env = simpy.Environment()
process_q = simpy.Store(env)
exit_q = simpy.Store(env)
env.process(gen_ents(env, process_q))
env.process(batch_process(env, process_q, exit_q, 4))
env.run(1000)
print(f'Simulation has finish a time {env.now}')
我正在为一个非常长的多次重入制造过程开发一个离散偶数模拟器（零件按特定顺序经过多个工具，并且经常多次返回到同一个工具）。该程序的整体流程运行良好，并在各种构建计划场景下给出了预期的队列深度、拥塞点等。我还没有找到解决方法的挑战是容量 >1 的工具可能 运行 ，一次多个部分，但它必须同时启动它们（即它们可能 运行 在房间或浴缸中，例如当一个部分是 运行 时不能 opened/accessed 添加另一个）。
因此，我正在寻找一种方法来实现一个工具，比如 capacity=4，这样在 运行 开始时，如果队列中有 4 个或更多项目，它'我将加载其中的 4 个和 运行 它们，但是如果在时间为零时只有一个部分，则该部分 运行s 以及进入队列的任何东西，而该部分是 运行 ning要等到运行结束
我的代码太长太复杂，无法说明问题，但是著名的简单“加油站”示例（代码如下）很好地描述了问题。这段代码中表达了我试图摆脱的行为。也就是说，加油站的容量为 2，一辆车出现并占据了其中一个槽位，然后一段时间后另一辆车到达并占据了剩余的槽位。这对加油站来说很棒，但我试图阻止以后的用户在 运行 启动后获得访问权限。
我可以设想为该工具提供一个 属性，如 self.status，并在使用该工具时将状态设置为 'busy' 或类似的东西，或者可能使用 self.res.users 以某种方式做到这一点，但我想知道是否有一些更简单的方式让我的工具以所需的方式运行。
谢谢！
import itertools
import random
import simpy
RANDOM_SEED = 42
GAS_STATION_SIZE = 200 # liters
THRESHOLD = 10 # Threshold for calling the tank truck (in %)
FUEL_TANK_SIZE = 50 # liters
FUEL_TANK_LEVEL = [5, 25] # Min/max levels of fuel tanks (in liters)
REFUELING_SPEED = 2 # liters / second
TANK_TRUCK_TIME = 300 # Seconds it takes the tank truck to arrive
T_INTER = [30, 300] # Create a car every [min, max] seconds
SIM_TIME = 1000 # Simulation time in seconds
def car(name, env, gas_station, fuel_pump):
"""A car arrives at the gas station for refueling.
It requests one of the gas station's fuel pumps and tries to get the
desired amount of gas from it. If the stations reservoir is
depleted, the car has to wait for the tank truck to arrive.
"""
fuel_tank_level = random.randint(*FUEL_TANK_LEVEL)
print('%s arriving at gas station at %.1f' % (name, env.now))
with gas_station.request() as req:
start = env.now
# Request one of the gas pumps
yield req
# Get the required amount of fuel
liters_required = FUEL_TANK_SIZE - fuel_tank_level
yield fuel_pump.get(liters_required)
# The "actual" refueling process takes some time
yield env.timeout(liters_required / REFUELING_SPEED)
print('%s finished refueling in %.1f seconds.' % (name,
env.now - start))
def gas_station_control(env, fuel_pump):
"""Periodically check the level of the *fuel_pump* and call the tank
truck if the level falls below a threshold."""
while True:
if fuel_pump.level / fuel_pump.capacity * 100 < THRESHOLD:
# We need to call the tank truck now!
print('Calling tank truck at %d' % env.now)
# Wait for the tank truck to arrive and refuel the station
yield env.process(tank_truck(env, fuel_pump))
yield env.timeout(10) # Check every 10 seconds
def tank_truck(env, fuel_pump):
"""Arrives at the gas station after a certain delay and refuels it."""
yield env.timeout(TANK_TRUCK_TIME)
print('Tank truck arriving at time %d' % env.now)
amount = fuel_pump.capacity - fuel_pump.level
print('Tank truck refuelling %.1f liters.' % amount)
yield fuel_pump.put(amount)
def car_generator(env, gas_station, fuel_pump):
"""Generate new cars that arrive at the gas station."""
for i in itertools.count():
yield env.timeout(random.randint(*T_INTER))
env.process(car('Car %d' % i, env, gas_station, fuel_pump))
# Setup and start the simulation
print('Gas Station refuelling')
random.seed(RANDOM_SEED)
# Create environment and start processes
env = simpy.Environment()
gas_station = simpy.Resource(env, 2)
fuel_pump = simpy.Container(env, GAS_STATION_SIZE, init=GAS_STATION_SIZE)
env.process(gas_station_control(env, fuel_pump))
env.process(car_generator(env, gas_station, fuel_pump))
# Execute!
env.run(until=SIM_TIME)
所以在我看来，每辆车都抓住了一个泵，但你真正想要的是一个泵一次最多能抓住两辆车。所以你需要一个进行批处理的过程。这是一个快速而肮脏的批处理程序。我为输入队列使用一个存储，当我的每个 yield get() 获得第一个实体时，我检查队列是否有更多实体要处理，这个 yield 真正做的是导致我的进程在队列是空
"""
A simple example of a process that process entities in batches
Programmer: Michael R. Gibbs
"""
import simpy
import random
def batch_process(env, ent_q, next_q, max_cap):
"""
grabs up to max_cap of entites and processes them
"""
while True:
# use a yield to wait when queue is empty
ent_1 = yield ent_q.get()
# grabs up to capacity -1 more
p_batch = [ent_1] + ent_q.items[:(max_cap - 1)]
ent_q.items = ent_q.items[(max_cap - 1):]
print(f'{env.now:.2f} grabbed {len(p_batch)} entites leaving {len(ent_q.items)} left in queue')
# do the processing
yield env.timeout(10)
# send entitles to next stop
next_q.items = next_q.items + p_batch
def gen_ents(env, q):
"""
Generate the arrival of entities
arrival profile starts slow
has a peek
then nothing which drains the queue
and puts the process in a wait mode
another peek
"""
for _ in range(10):
yield env.timeout(random.uniform(1,6))
q.put(object())
for _ in range(15):
yield env.timeout(random.uniform(1,3))
q.put(object())
yield env.timeout(50)
for _ in range(15):
yield env.timeout(random.uniform(1,3))
q.put(object())
for _ in range(10):
yield env.timeout(random.uniform(1,7))
q.put(object())
# boot up model
random.seed(100)
env = simpy.Environment()
process_q = simpy.Store(env)
exit_q = simpy.Store(env)
env.process(gen_ents(env, process_q))
env.process(batch_process(env, process_q, exit_q, 4))
env.run(1000)
print(f'Simulation has finish a time {env.now}')