如何在Python Simpy中实现跨类连续流程调度？

核心思路 #

• 让各个独立模块（切割、运输、生成器）互相持有后续环节的实例引用，实现流程触发
• 为运输环节设置等待区，用Simpy的Store处理运输繁忙时的排队逻辑

完整实现代码 #

import simpy

# 切割机器类
class CuttingMachine:
    def __init__(self, env, num_machines):
        self.env = env
        self.machines = simpy.Resource(env, num_machines)
        self.transport_system = None  # 后续运输系统的引用

    def set_transport_system(self, transport_system):
        self.transport_system = transport_system

    def cut(self, material):
        # 申请切割机器
        with self.machines.request() as req:
            yield req
            print(f"物料{material.id}开始切割，时间：{self.env.now:.2f}")
            # 模拟切割耗时，可按需调整
            yield self.env.timeout(2)
            print(f"物料{material.id}切割完成，时间：{self.env.now:.2f}")
            # 触发运输流程
            self.env.process(self.transport_system.transport_material(material))

# 运输系统类
class TransportSystem:
    def __init__(self, env, num_operators):
        self.env = env
        self.operators = simpy.Resource(env, num_operators)
        self.waiting_area = simpy.Store(env)  # 运输等待区
        # 启动运输调度进程
        self.env.process(self.dispatch_transport())

    def transport_material(self, material):
        # 物料进入运输等待区
        yield self.waiting_area.put(material)
        print(f"物料{material.id}进入运输等待区排队，时间：{self.env.now:.2f}")

    def dispatch_transport(self):
        while True:
            # 从等待区取出待运输物料
            material = yield self.waiting_area.get()
            # 申请空闲操作员
            with self.operators.request() as req:
                yield req
                print(f"操作员接手物料{material.id}，开始运输，时间：{self.env.now:.2f}")
                # 模拟运输到下一工序的时间
                yield self.env.timeout(1.5)
                print(f"物料{material.id}送达下一工序，时间：{self.env.now:.2f}")
                # 模拟操作员返程时间
                yield self.env.timeout(1.5)
                print(f"操作员返回待命，时间：{self.env.now:.2f}")

# 物料生成器类
class MaterialGenerator:
    def __init__(self, env, cutting_machine):
        self.env = env
        self.cutting_machine = cutting_machine
        self.next_material_id = 1
        # 启动物料生成进程
        self.env.process(self.generate_materials())

    def generate_materials(self):
        while True:
            # 每8.8分钟生成一个物料
            yield self.env.timeout(8.8)
            material = type('Material', (object,), {'id': self.next_material_id})()
            print(f"生成新物料{material.id}，时间：{self.env.now:.2f}")
            self.next_material_id += 1
            # 触发切割流程
            self.env.process(self.cutting_machine.cut(material))

# 启动仿真
if __name__ == "__main__":
    env = simpy.Environment()
    # 初始化各模块
    cutter = CuttingMachine(env, num_machines=2)
    transporter = TransportSystem(env, num_operators=1)
    # 建立模块间的关联
    cutter.set_transport_system(transporter)
    # 启动物料生成
    MaterialGenerator(env, cutter)
    # 运行仿真60分钟
    env.run(until=60)

关键细节说明 #

• 模块关联：通过set_transport_system方法让切割机器获取运输系统的实例，切割完成后直接调用运输方法，解决跨类调用问题
• 运输等待区：Simpy.Store自动处理物料排队，运输系统的调度进程持续从等待区取货，无需手动管理队列
• 资源申请：切割机器和操作员都用Simpy.Resource管理，自动处理资源空闲等待逻辑

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