
本文将深入探讨如何使用python的simpy库构建一个离散事件模拟模型,以模拟工厂生产线的复杂流程。我们将详细介绍资源管理、零件生成、多阶段加工以及如何通过simpy的`priorityresource`和事件机制来精确控制生产流程和资源分配,旨在提供一个清晰、专业的simpy应用指南。
离散事件模拟(Discrete Event Simulation, DES)是一种强大的建模工具,用于分析复杂系统随时间变化的动态行为。在制造业中,DES常用于优化生产线布局、评估资源利用率、识别瓶颈以及预测系统性能。SimPy是一个基于Python的DES框架,它通过提供进程(processes)和资源(resources)等核心概念,使得构建这类模拟模型变得直观和高效。
在工厂生产线的场景中,零件(实体)在不同的工位(资源)之间移动,并经历一系列加工步骤(事件)。这些工位可能包括操作员、机器人、夹具等,它们通常数量有限,需要被零件请求和释放。SimPy能够模拟这些实体与资源的交互,以及加工过程所需的时间,从而揭示生产线的整体运作效率。
在构建SimPy模型时,首先定义一些全局常量和模拟环境是良好的实践。
我们通常会将模拟中不变的参数(如加工时间、资源数量)集中管理,以提高代码的可读性和可维护性。
import simpy
import itertools
class g: # 全局常量类
t_sim = 600 # 模拟最大时长,例如600秒 (原问题为一天,这里为演示缩短)
operator_qty = 2 # 操作员数量
sWeld_t = 2.75 # 点焊时间(秒/次)
Adh_t = 1/60 # 涂胶时间(秒/毫米)
ArcStud_t = 5.5 # 拉弧焊时间(秒/次)
ProjStud_t = 10 # 凸焊时间(秒/次)g 类封装了所有与模拟过程相关的固定参数,例如模拟总时长t_sim、操作员数量operator_qty,以及各种加工步骤的单位时间。
cell_model 类代表了整个工厂单元,负责创建SimPy环境、定义所有可用资源,并启动零件生成器。
class cell_model: # 代表整个工厂单元
def __init__(self): # 单元的初始条件
self.env = simpy.Environment() # 创建SimPy环境
self.part_counter = 0 # 零件计数器
# 声明所有资源,使用PriorityResource以支持优先级调度
# 操作员资源,可被多个进程共享
self.OP = simpy.PriorityResource(self.env, capacity=g.operator_qty)
# 机器人资源,每个机器人一次只能处理一个零件
self.R1 = simpy.PriorityResource(self.env, capacity=1)
self.R2 = simpy.PriorityResource(self.env, capacity=1)
self.R3 = simpy.PriorityResource(self.env, capacity=1)
self.R4 = simpy.PriorityResource(self.env, capacity=1)
# 夹具资源,每个夹具一次只能放置一个零件
self.Fix01 = simpy.PriorityResource(self.env, capacity=1)
self.Fix02 = simpy.PriorityResource(self.env, capacity=1)
self.Fix101 = simpy.PriorityResource(self.env, capacity=1)
# 启动零件生成器进程
self.env.process(self.part_generator())在 __init__ 方法中:
make_part 函数是模拟中最重要的部分,它描述了一个零件从进入生产线到完成所有工序的完整路径。每个零件都将作为独立的SimPy进程通过这个流程。
def make_part(self, p_Id): # 描述单个零件在单元中的所有流程
# 1. 操作员将零件带到工作台
OP_req1 = self.OP.request(priority=p_Id)
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求操作员")
yield OP_req1 # 请求操作员并等待其可用
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用操作员")
yield self.env.timeout(29) # 操作员移动零件、贴胶、放置到夹具01
# 2. 请求夹具01
Fix01_req = self.Fix01.request(priority=p_Id)
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求夹具01")
yield Fix01_req # 等待夹具01可用
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用夹具01")
yield self.env.timeout(27) # 操作员加载零件,退出
# 3. 释放操作员
self.OP.release(OP_req1) # 操作员完成任务,释放回资源池
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放操作员")
yield self.env.timeout(0) # 短暂暂停
# 4. 机器人R2进行点焊 (使用with语句自动释放)
with self.R2.request(priority=p_Id) as R2_req:
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求机器人R2")
yield R2_req # 请求机器人R2
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用机器人R2 (第一阶段)")
yield self.env.timeout(g.sWeld_t * 11) # R2进行11次点焊
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放机器人R2 (with块自动释放)")
# 5. 机器人R1移动零件并进行焊接
R1_req = self.R1.request(priority=p_Id)
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求机器人R1")
yield R1_req # 请求机器人R1
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用机器人R1")
yield self.env.timeout(8) # R1将零件移出夹具01
print(f"零件 {p_Id} 在 {self.env.now:.2f} 离开夹具01")
# 6. 释放夹具01
self.Fix01.release(Fix01_req) # 夹具01清空,释放回资源池
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放夹具01")
yield self.env.timeout(g.sWeld_t * 13) # R1进行13次点焊
yield self.env.timeout(g.ArcStud_t * 5) # R1进行5次拉弧焊
yield self.env.timeout(8) # R1移动到夹具101
# 7. 请求夹具101
Fix101_req = self.Fix101.request(priority=p_Id)
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求夹具101")
yield Fix101_req # 请求夹具101
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用夹具101")
print(f"零件 {p_Id} 在 {self.env.now:.2f} 进入夹具101")
# 8. 释放机器人R1
self.R1.release(R1_req) # R1完成任务,释放回资源池
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放机器人R1")
# 9. 机器人R4进行拉弧焊 (使用with语句自动释放)
with self.R4.request(priority=p_Id) as R4_req:
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求机器人R4")
yield R4_req # 请求机器人R4
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用机器人R4")
yield self.env.timeout(g.ArcStud_t * 6) # R4进行6次拉弧焊
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放机器人R4")
# 10. 机器人R3移动零件并进行焊接、涂胶
R3_req = self.R3.request(priority=p_Id)
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求机器人R3")
yield R3_req # 请求机器人R3
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用机器人R3")
yield self.env.timeout(8) # R3将零件移出夹具101
print(f"零件 {p_Id} 在 {self.env.now:.2f} 离开夹具101")
# 11. 释放夹具101
self.Fix101.release(Fix101_req) # 夹具101清空,释放回资源池
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放夹具101")
yield self.env.timeout(g.ProjStud_t * 6) # R3进行6次凸焊
yield self.env.timeout(225 * g.Adh_t) # 涂255mm胶
print(f"零件 {p_Id} 在 {self.env.now:.2f} 完成涂胶")
yield self.env.timeout(8) # R3移动到夹具02
# 12. 请求夹具02
Fix02_req = self.Fix02.request(priority=p_Id)
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求夹具02")
yield Fix02_req # 请求夹具02
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用夹具02")
print(f"零件 {p_Id} 在 {self.env.now:.2f} 进入夹具02")
# 13. 释放机器人R3
self.R3.release(R3_req) # R3完成任务,释放回资源池
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放机器人R3")
# 14. 操作员进行第二阶段操作 (使用with语句自动释放)
with self.OP.request(priority=p_Id) as OP_req2:
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求操作员 (第二阶段)")
yield OP_req2 # 请求操作员
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用操作员 (第二阶段)")
yield self.env.timeout(38) # 操作员进行第二阶段操作
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放操作员 (第二阶段)")
# 15. 机器人R2再次进行焊接 (使用with语句自动释放)
with self.R2.request(priority=p_Id) as R2_req2:
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求机器人R2 (第二阶段)")
yield R2_req2 # 请求机器人R2
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用机器人R2 (第二阶段)")
print(f"零件 {p_Id} 在 {self.env.now:.2f} 机器人R2移动到夹具02工作")
yield self.env.timeout(g.sWeld_t * 35) # R2进行35次点焊
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放机器人R2 (第二阶段)")
# 16. 操作员移除零件并完成流程 (使用with语句自动释放)
with self.OP.request(priority=p_Id) as OP_req3:
print(f"零件 {p_Id} 在 {self.env.now:.2f} 请求操作员 (第三阶段)")
yield OP_req3 # 请求操作员
print(f"零件 {p_Id} 在 {self.env.now:.2f} 占用操作员 (第三阶段)")
yield self.env.timeout(2) # 操作员移除零件
# 17. 释放夹具02
self.Fix02.release(Fix02_req) # 夹具02清空,释放回资源池
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放夹具02")
print(f"零件 {p_Id} 在 {self.env.now:.2f} 离开生产线")
print(f"零件 {p_Id} 在 {self.env.now:.2f} 释放操作员 (第三阶段)")part_generator 进程负责在模拟过程中不断创建新的零件。
def part_generator(self):
for p_Id in itertools.count(): # 无限生成零件ID
yield self.env.timeout(1) # 每隔1秒生成一个新零件
self.env.process(self.make_part(p_Id)) # 启动一个新零件的加工进程itertools.count() 会生成一个无限递增的整数序列作为 p_Id。yield self.env.timeout(1) 表示每隔1个模拟时间单位(秒),就会生成一个新的零件。self.env.process(self.make_part(p_Id)) 则为每个新零件启动一个独立的 make_part 进程。
run 方法是启动整个模拟的入口。
def run(self, t_sim=g.t_sim): # 运行方法启动实体生成器并告诉Simpy运行环境
self.env.run(until=t_sim) # 运行模拟直到指定时间
# 运行模拟
print("Simulation Starting...")
my_model = cell_model()
my_model.run()
print("Simulation Finished.")my_model.run() 调用 self.env.run(until=t_sim),这将启动SimPy环境,并执行所有已注册的进程,直到模拟时间达到 t_sim。
以上就是使用SimPy进行工厂生产线离散事件模拟教程的详细内容,更多请关注php中文网其它相关文章!
每个人都需要一台速度更快、更稳定的 PC。随着时间的推移,垃圾文件、旧注册表数据和不必要的后台进程会占用资源并降低性能。幸运的是,许多工具可以让 Windows 保持平稳运行。
Copyright 2014-2025 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号