基于SimPy的工厂流水线建模及作业队列处理咨询

问题描述 #

• 刚接触SimPy，尝试建模工厂车间场景：零件从料箱流出后经过多道工序（Robot 2会在工序首尾两次被调用），期望零件按前工位空闲顺序进入流水线。
• 当前实现通过创建多个并行make_part进程，用PriorityResource保证零件按进入顺序流出，但认为未准确模拟目标系统，希望获取将作业队列通过单一流程处理的示例及相关建议。

当前实现代码 #

import simpy
import itertools

class g:#global constants
    t_sim = 24*60*60 #maximum simulation set to a day and 1 second
    
    operator_qty = 2 #how many operators? (assume as interchangeable across any operator type process, helps us ask "what is the difference between two running it and one?")
    
    sWeld_t = 2.75 #seconds per spot weld
    Adh_t = 1/60 #seconds/mm to apply sealant or adhesive
    ArcStud_t = 5.5 #seconds/drawn arc stud weld
    ProjStud_t = 10 #seconds/projection stud weld

        
class cell_model: #represents entire factory cell in which all processes, resources and the modelling environment itself are held
    def __init__(self):#initial conditions of cell
        self.env = simpy.Environment() #create simpy environment
        self._dict=dict()
        self.part_counter = 0 #initialize @ part number zero
        self.OP = simpy.PriorityResource(self.env,capacity=g.operator_qty)#implement resources, limited number of operators to do operator process, we assume operators can switch
        #Declare all robots as resources, process cannot proceed unless proper resource is available
        self.R1 = simpy.PriorityResource(self.env,capacity=1) #declare each robot as a resource (like a bin), each robot can only hold 1 part entity at a time, each robot can be in or out of a process (availability changes by request and release commands within a process)
        self.R2 = simpy.PriorityResource(self.env,capacity=1)
        self.R3 = simpy.PriorityResource(self.env,capacity=1)
        self.R4 = simpy.PriorityResource(self.env,capacity=1)
        #Declare all Fixture points as resources, a fixture can be interacted with as long as the robot is outside the relevant fixture process
        self.Fix01 = simpy.PriorityResource(self.env,capacity=1)#a bin of capacity 01, holds one part at a time, becomes available once released by a process, becomes unavailable when requested by a process
        self.Fix02 = simpy.PriorityResource(self.env,capacity=1)
        self.Fix101 = simpy.PriorityResource(self.env,capacity=1)

        self.env.process(self.part_generator())
   
            
    def make_part(self,p_Id):#describe the set of processes one single part would experience going through the cell, we will then try to cram the parts throught this process as fast as possible
        with self.OP.request(priority=p_Id) as OP:
            yield OP #request an operator and wait until one is available
            print("Part ", p_Id, " is next available at ", self.env.now, sep="")#report part in time
            yield self.env.timeout(29)#operator walks part to tape bench, tapes part, takes part to fixture 01
        with self.Fix01.request(priority=p_Id) as Fix01:
            yield Fix01 #while using the operator resource, request the Fixture, wait until it is avialable to proceed further
            yield self.env.timeout(27) #operator loads parts, exits, hits start.
            yield self.OP.release(OP) #done with operator, release him back to resource pool
            print("Operator 01 released part ", p_Id, " at ", self.env.now, sep="")
            yield self.env.timeout(0) #take a beat b4 next request
        with self.R2.request(priority=p_Id) as R2:    
            yield R2 #I need R2 now, request to proceed
            print("R2 moves to Fixture 01 to work on part ", p_Id, " at ", self.env.now, sep="")
            yield self.env.timeout(g.sWeld_t*11) #timeout, R2 11 welds @2.75s/weld
            yield self.R2.release(R2) #weld process done, release the R2 resource back to pool
        with self.R1.request(priority=p_Id) as R1:
            yield R1 #request R1 to move part out of fixture 01
            yield self.env.timeout(8) #R1 move part out of Fix01
            print("Part ", p_Id, " is leaving Fixture 01 at ", self.env.now, sep="")
            yield self.Fix01.release(Fix01) #fixture01 is clear, release back to resource pool
            yield self.env.timeout(g.sWeld_t*13) #do 13 welds
            yield self.env.timeout(0)#take a beat
            yield self.env.timeout(g.ArcStud_t*5) #do 5 drawn arc stud 
            yield self.env.timeout(8) #R1 travels to Fixture 101 drop off
        with self.Fix101.request(priority=p_Id) as Fix101:
            yield Fix101 #ask for Fix101 resource to place part into
            print("Part ", p_Id, " enters Fixture 101 at ", self.env.now, sep="")
            yield self.R1.release(R1) #done with R1 robot, send back to resource pool
        with self.R4.request(priority=p_Id) as R4:
            yield R4 #I need R4 to proceed off of Fixture 101 drop off points
            yield self.env.timeout(g.ArcStud_t*6) #R4 does arc_studs within fixture 101 drop off
            yield self.R4.release(R4) #done with R4, release back to resource pool
        with self.R3.request(priority=p_Id) as R3:
            yield R3 #wait for R3 to be available to move part out of fixture 101 drop off
            yield self.env.timeout(8) #R3 move part out of fix101, thus freeing it
            print("Part ", p_Id, " leaves Fixture 101 at ", self.env.now, sep="")
            yield self.Fix101.release(Fix101) #release fixture101 drop off back to resource pool
            yield self.env.timeout(g.ProjStud_t*6) #do 6 projection studs
            yield self.env.timeout(0) #move to next station
            yield self.env.timeout(225*g.Adh_t)#apply 255mm of adhesive/sealant
            print("Part ", p_Id, " has adhesive stage finished at ", self.env.now, sep="")
            yield self.env.timeout(8)#R3 travels to fixture 02 for drop off
        with self.Fix02.request(priority=p_Id) as Fix02:
            yield Fix02#request the Fixture 02 bin to place the part into
            yield self.env.timeout(0)#take a beat
            print("Part ", p_Id, " enters Fixture 02 at ", self.env.now, sep="")
            yield self.R3.release(R3)#done with R3, release to resource pool
        with self.OP.request(priority=p_Id) as OP:
            yield OP#now need an operator to load more parts
            yield self.env.timeout(38) #take 38 seconds to do Operator 02 process
            yield self.OP.release(OP)#done with operator, send back to resource pool
            yield self.env.timeout(0)#take a beat
        with self.R2.request(priority=p_Id) as R2:
            yield R2#request the R2 to finish the welding on this part
            print("R2 moves to Fixture 02 to work on part ", p_Id, " at ", self.env.now, sep="")
            yield self.env.timeout(g.sWeld_t*35)#do 35 welds
            yield self.R2.release(R2)#done with R2, send back to resource pool
            yield self.env.timeout(0)#take a beat
        with self.OP.request(priority=p_Id) as OP:
            yield OP#need an operator to remove the part
            yield self.env.timeout(2)#operator removes part and restarts cycle
            yield self.Fix02.release(Fix02)#done with Fixture 02, release back to pool of resources
            print("Part ", p_Id, " leaves process at ", self.env.now, sep="")#report part out time
            yield self.OP.release(OP)#release operator back to resource pool 
            yield self.env.timeout(0)#take a beat
  
    def part_generator(self):
        for p_Id in itertools.count():
            yield self.env.timeout(1)
            self.env.process(self.make_part(p_Id)) 
       
    def run(self,t_sim=g.t_sim): #run method starts up entity generators and tells Simpy to start running the environment for a duration specified in class g:
        self.env.run(until=t_sim)



#run the simulation
print("Simulation Starting...")
my_model=cell_model()
my_model.run()
print()

优化方案与示例 #

1. 单一队列+工位独立流程的核心思路 #

放弃每个零件对应一个make_part进程的模式，改用状态驱动的队列管理：

• 用FilterStore维护全局零件队列，零件通过状态字段标记当前待处理的工序
• 每个工位（如胶带处理、Fixture01加载、R2焊接等）作为独立的SimPy进程，循环从队列中筛选出符合自身处理条件的零件，完成加工后更新零件状态并放回队列，供下一个工位处理
• 这种模式天然实现“前工位空闲顺序”：哪个工位先空闲，就先从队列中取对应状态的零件处理

2. 关键代码实现示例 #

第一步：定义零件类，跟踪状态

class Part:
    def __init__(self, part_id):
        self.id = part_id
        self.status = "waiting_for_tape"  # 初始状态：等待操作员胶带处理
        # 可扩展：记录各工序完成时间、加工参数等

第二步：重构cell_model，用队列管理零件

class cell_model:
    def __init__(self):
        self.env = simpy.Environment()
        # 用FilterStore存储零件，支持按状态筛选
        self.part_queue = simpy.FilterStore(self.env)
        # 资源改用普通Resource，无需优先级（队列筛选逻辑替代优先级控制）
        self.OP = simpy.Resource(self.env, capacity=g.operator_qty)
        self.R1 = simpy.Resource(self.env, capacity=1)
        self.R2 = simpy.Resource(self.env, capacity=1)
        self.R3 = simpy.Resource(self.env, capacity=1)
        self.R4 = simpy.Resource(self.env, capacity=1)
        self.Fix01 = simpy.Resource(self.env, capacity=1)
        self.Fix02 = simpy.Resource(self.env, capacity=1)
        self.Fix101 = simpy.Resource(self.env, capacity=1)

        # 启动零件生成器和所有工位处理流程
        self.env.process(self.part_generator())
        self.env.process(self.operator_tape_process())
        self.env.process(self.fix01_load_process())
        self.env.process(self.r2_fix01_weld_process())
        self.env.process(self.r1_move_fix01_process())
        # 后续工位流程可按同样逻辑添加

    def part_generator(self):
        part_id = 0
        while True:
            yield self.env.timeout(1)
            new_part = Part(part_id)
            self.part_queue.put(new_part)
            print(f"Part {new_part.id} generated at {self.env.now:.2f}")
            part_id += 1

    # 工位1：操作员胶带处理
    def operator_tape_process(self):
        while True:
            # 筛选出等待胶带处理的零件
            part = yield self.part_queue.get(lambda p: p.status == "waiting_for_tape")
            with self.OP.request() as req:
                yield req
                print(f"Part {part.id} starts tape process at {self.env.now:.2f}")
                yield self.env.timeout(29)
                # 更新零件状态，放回队列
                part.status = "waiting_for_fix01_load"
                self.part_queue.put(part)
                print(f"Part {part.id} finishes tape process at {self.env.now:.2f}")

    # 工位2：Fixture01加载（需要操作员+夹具）
    def fix01_load_process(self):
        while True:
            part = yield self.part_queue.get(lambda p: p.status == "waiting_for_fix01_load")
            # 同时请求操作员和夹具资源，避免死锁
            with self.OP.request() as op_req, self.Fix01.request() as fix_req:
                yield op_req & fix_req
                print(f"Part {part.id} starts loading to Fix01 at {self.env.now:.2f}")
                yield self.env.timeout(27)
                part.status = "waiting_for_r2_weld"
                self.part_queue.put(part)
                print(f"Part {part.id} loaded to Fix01 at {self.env.now:.2f}")

    # 工位3：R2在Fixture01焊接
    def r2_fix01_weld_process(self):
        while True:
            part = yield self.part_queue.get(lambda p: p.status == "waiting_for_r2_weld")
            with self.R2.request() as req:
                yield req
                print(f"R2 starts welding Part {part.id} at {self.env.now:.2f}")
                yield self.env.timeout(g.sWeld_t * 11)
                part.status = "waiting_for_r1_move_fix01"
                self.part_queue.put(part)
                print(f"R2 finishes welding Part {part.id} at {self.env.now:.2f}")

3. 额外建议 #

• 避免不必要的PriorityResource：如果不需要严格按零件ID优先级处理，普通Resource的FIFO队列配合FilterStore的状态筛选，更贴合“前工位空闲顺序”的需求
• 资源请求原子性：对于需要多个资源的工序（如加载Fixture），用&同时请求所有资源，防止出现占用部分资源导致死锁的情况
• 状态扩展：可以给零件添加更多状态字段（如当前所在工位、加工异常标记等），让模型更贴近真实生产场景
• 日志与统计：将打印日志改为统一的记录函数，方便后续导出加工时间、资源利用率等统计数据

内容的提问来源于stack exchange，提问作者Frosty1542