使用SimPy构建复杂工厂流程的离散事件仿真模型-Python教程-PHP中文网

使用SimPy构建复杂工厂流程的离散事件仿真模型

本教程详细阐述如何使用python的simpy库构建一个复杂的工厂装配线离散事件仿真模型。我们将探讨simpy环境、进程和资源的核心概念，重点解决资源管理中的常见陷阱，特别是`with`语句与显式`request/release`方法的选择。通过优化代码示例，文章将展示如何精确模拟零件在多阶段、多资源共享环境中的流动，确保资源正确分配与释放，并讨论如何利用`priorityresource`维持生产顺序，以及在仿真过程中识别和解决潜在的死锁问题。

引言：SimPy与工厂流程建模

离散事件仿真（DES）是一种强大的工具，用于分析和优化复杂系统，如生产线、物流系统或服务中心。SimPy是一个基于Python的离散事件仿真库，它允许用户通过定义“进程”（processes）来描述系统中的动态行为，这些进程通过“资源”（resources）进行交互和竞争。在工厂流程建模中，零件、机器人、操作员和工位（夹具）都可以被视为SimPy中的实体和资源。精确地管理这些资源的请求与释放是构建有效仿真模型的关键。

SimPy核心概念回顾

在深入工厂模型之前，我们先回顾SimPy的几个核心概念：

simpy.Environment(): 仿真环境，负责管理时间和调度事件。所有进程都在此环境中运行。
env.process(generator): 将一个生成器函数（或协程）注册为一个SimPy进程。进程是SimPy中执行活动的实体。
yield env.timeout(delay): 进程暂停执行delay时间单位，然后继续。这是SimPy中模拟时间流逝的基本方式。
simpy.Resource(env, capacity=N): 一种通用资源，可以被N个进程同时使用。如果资源被占用，后续请求的进程会进入等待队列。
simpy.PriorityResource(env, capacity=N): 带有优先级的资源。请求该资源的进程可以指定一个优先级（数字越小优先级越高），优先级高的进程将优先获得资源。

原始模型的问题分析

在SimPy中，资源的请求和释放是核心操作。通常，有两种方式来管理资源：

显式请求与释放: request = resource.request(priority=p); yield request; ...; yield resource.release(request)
使用with语句: with resource.request(priority=p) as request: yield request; ...

with语句是Python的上下文管理器，它确保在代码块结束时自动释放资源。这在许多场景下非常方便，但对于某些复杂的仿真逻辑，它可能导致问题。

原始代码中，许多资源（如操作员OP、机器人R1、夹具Fix01）被错误地放置在with语句中。例如：

with self.OP.request(priority=p_Id) as OP:
    yield OP # request an operator
    # ... operator performs task 1 ...
    yield self.Fix01.request(priority=p_Id) as Fix01: # Nested with, problematic
        yield Fix01
        # ... operator performs task 2 ...
        yield self.OP.release(OP) # This release is inside Fix01's with block, but OP's with block hasn't ended.

登录后复制

这里的主要问题是：

过早释放: with语句会在其代码块执行完毕后立即释放资源。在工厂流程中，一个操作员可能在完成一个任务后，需要等待另一个资源（如夹具）可用，然后才能继续下一个任务，而在此期间，该操作员仍然处于被占用状态，不能被其他零件请求。如果使用with语句，操作员可能会在第一个任务结束后就被自动释放，这与实际的物理流程不符。
资源持有跨越多个不连续步骤: 某些资源（如操作员或机器人）可能在多个不连续的生产阶段中被同一个零件占用，中间可能穿插其他资源的操作或长时间的等待。with语句不适合这种长时间、非连续的资源持有。
嵌套with语句的复杂性: 嵌套的with语句会使资源管理逻辑变得复杂且容易出错，特别是当内部资源需要外部资源进行释放时。

正确的做法是，对于那些需要被长时间持有、或在不同时间点由不同实体释放的资源，应使用显式的request()和release()方法。只有当资源在特定的、连续的、短期的操作块内被请求和释放时，才适合使用with语句。

天工大模型

中国首个对标ChatGPT的双千亿级大语言模型

115

查看详情

优化后的资源管理策略

为了准确模拟工厂流程，我们必须采用更精细的资源管理策略：

显式请求与释放 (request() / release()): 当一个资源需要被一个进程长时间占用，或者在被占用期间需要与其他资源进行复杂的交互，或者其释放点与请求点在代码逻辑上相距较远时，应使用显式的方法。
- 首先调用 resource.request(priority=p_Id) 来创建一个请求对象。
- 然后 yield request 等待资源可用并被分配。
- 当资源不再需要时，调用 resource.release(request) 来释放资源。这种方式赋予了开发者对资源生命周期的完全控制权。
上下文管理器 (with 语句): 当一个资源仅在一个明确定义的、相对独立的、短期的操作块内被请求和释放时，with语句依然是简洁高效的选择。例如，机器人执行一段连续的焊接操作，完成后立即释放。
优先级资源 (PriorityResource): 在工厂生产线中，通常希望零件按照它们进入系统的顺序进行处理（即“先到先服务”）。PriorityResource允许我们为每个请求指定一个优先级。通过将零件ID作为优先级（例如，p_Id），可以确保编号较小的零件（即先生成的零件）在资源竞争中具有更高的优先级，从而维持生产顺序。

SimPy工厂模型实现详解

现在，我们根据上述优化策略来构建一个更健壮的SimPy工厂模型。

1. 全局常量与初始化 (g 类与 cell_model.init)

g类定义了仿真所需的全局常量，如仿真时长、操作员数量、各种操作的时间等。cell_model类初始化了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 # 凸焊时间

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()) # 启动零件生成器

登录后复制

2. 零件生产流程 (make_part 方法)

make_part方法描述了单个零件从进入系统到完成所有工序的完整路径。这是模型的核心，包含了所有资源请求、操作时间和释放逻辑。

    def make_part(self, p_Id): # 描述单个零件的生产流程
        # --- 阶段1：操作员处理并放置到夹具Fix01 ---
        OP_req1 = self.OP.request(priority=p_Id) # 显式请求操作员
        print(f"part {p_Id} OP request at {self.env.now}")
        yield OP_req1 # 等待操作员可用
        print(f"part {p_Id} OP seized at {self.env.now}")
        print("Part ", p_Id, " is next available at ", self.env.now, sep="")
        yield self.env.timeout(29) # 操作员移动零件到工作台，处理，再移动到夹具01

        Fix01_req = self.Fix01.request(priority=p_Id) # 显式请求夹具01
        print(f"part {p_Id} Fix01 request at {self.env.now}")
        yield Fix01_req # 等待夹具01可用
        print(f"part {p_Id} Fix01 seized at {self.env.now}")
        yield self.env.timeout(27) # 操作员加载零件，离开，启动

        yield self.OP.release(OP_req1) # 操作员完成第一阶段任务，释放
        print(f"part {p_Id} OP released at {self.env.now}")
        print("Operator 01 released part ", p_Id, " at ", self.env.now, sep="")
        yield self.env.timeout(0) # 短暂暂停

        # --- 阶段2：机器人R2在夹具Fix01上进行点焊 (使用with语句，R2操作相对独立) ---
        with self.R2.request(priority=p_Id) as R2_req:
            print(f"part {p_Id} R2 request at {self.env.now}")
            yield R2_req # 请求R2
            print(f"part {p_Id} R2 seized at {self.env.now}")
            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) # R2执行11次点焊
        print(f"part {p_Id} R2 released at {self.env.now}") # R2自动释放

        # --- 阶段3：机器人R1移动零件，执行焊接，放置到夹具Fix101 ---
        R1_req = self.R1.request(priority=p_Id) # 显式请求R1
        print(f"part {p_Id} R1 request at {self.env.now}")
        yield R1_req # 等待R1可用
        print(f"part {p_Id} R1 seized at {self.env.now}")
        yield self.env.timeout(8) # R1将零件从夹具01移出
        print("Part ", p_Id, " is leaving Fixture 01 at ", self.env.now, sep="")
        yield self.Fix01.release(Fix01_req) # 夹具01被清空，释放
        print(f"part {p_Id} Fix01 released at {self.env.now}")

        yield self.env.timeout(g.sWeld_t * 13) # R1执行13次点焊
        yield self.env.timeout(0)
        yield self.env.timeout(g.ArcStud_t * 5) # R1执行5次弧焊
        yield self.env.timeout(8) # R1移动到夹具101放置点

        Fix101_req = self.Fix101.request(priority=p_Id) # 显式请求夹具101
        print(f"part {p_Id} Fix101 request at {self.env.now}")
        yield Fix101_req # 等待夹具101可用
        print(f"part {p_Id} Fix101 seized at {self.env.now}")
        print("Part ", p_Id, " enters Fixture 101 at ", self.env.now, sep="")
        yield self.R1.release(R1_req) # R1完成任务，释放
        print(f"part {p_Id} R1 released at {self.env.now}")

        # --- 阶段4：机器人R4在夹具Fix101上进行弧焊 (使用with语句) ---
        with self.R4.request(priority=p_Id) as R4_req:
            print(f"part {p_Id} R4 request at {self.env.now}")
            yield R4_req # 请求R4
            print(f"part {p_Id} R4 seized at {self.env.now}")
            yield self.env.timeout(g.ArcStud_t * 6) # R4执行6次弧焊
        print(f"part {p_Id} R4 released at {self.env.now}") # R4自动释放

        # --- 阶段5：机器人R3移动零件，执行凸焊、涂胶，放置到夹具Fix02 ---
        R3_req = self.R3.request(priority=p_Id) # 显式请求R3
        print(f"part {p_Id} R3 request at {self.env.now}")
        yield R3_req # 等待R3可用
        print(f"part {p_Id} R3 seized at {self.env.now}")
        yield self.env.timeout(8) # R3将零件从夹具101移出
        print("Part ", p_Id, " leaves Fixture 101 at ", self.env.now, sep="")
        yield self.Fix101.release(Fix101_req) # 夹具101被清空，释放
        print(f"part {p_Id} Fix101 released at {self.env.now}")

        yield self.env.timeout(g.ProjStud_t * 6) # R3执行6次凸焊
        yield self.env.timeout(0)
        yield self.env.timeout(225 * g.Adh_t) # R3执行涂胶
        print("Part ", p_Id, " has adhesive stage finished at ", self.env.now, sep="")
        yield self.env.timeout(8) # R3移动到夹具02放置点

        Fix02_req = self.Fix02.request(priority=p_Id) # 显式请求夹具02
        print(f"part {p_Id} Fix02 request at {self.env.now}")
        yield Fix02_req # 等待夹具02可用
        print(f"part {p_Id} Fix02 seized at {self.env.now}")
        yield self.env.timeout(0)
        print("Part ", p_Id, " enters Fixture 02 at ", self.env.now, sep="")
        yield self.R3.release(R3_req) # R3完成任务，释放
        print(f"part {p_Id} R3 released at {self.env.now}")

        # --- 阶段6：操作员处理，机器人R2完成焊接，操作员移除零件 ---
        with self.OP.request(priority=p_Id) as OP_req2: # 操作员进行第二阶段操作 (使用with语句)
            print(f"part {p_Id} OP request at {self.env.now}")
            yield OP_req2 # 请求操作员
            print(f"part {p_Id} OP seized at {self.env.now}")
            yield self.env.timeout(38) # 操作员执行操作02
        print(f"part {p_Id} OP released at {self.env.now}") # 操作员自动释放

        with self.R2.request(priority=p_Id) as R2_req2: # 机器人R2进行第二阶段焊接 (使用with语句)
            print(f"part {p_Id} R2 request at {self.env.now}")
            yield R2_req2 # 请求R2
            print(f"part {p_Id} R2 seized at {self.env.now}")
            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) # R2执行35次点焊
        print(f"part {p_Id} R2 released at {self.env.now}") # R2自动释放

        OP_req3 = self.OP.request(priority=p_Id) # 显式请求操作员移除零件
        print(f"part {p_Id} OP request at {self.env.now}")
        yield OP_req3 # 等待操作员可用
        print(f"part {p_Id} OP seized at {self.env.now}")
        yield self.env.timeout(2) # 操作员移除零件
        yield self.Fix02.release(Fix02_req) # 夹具02被清空，释放
        print(f"part {p_Id} Fix02 released at {self.env.now}")
        print("Part ", p_Id, " leaves process at ", self.env.now, sep="")
        yield self.OP.release(OP_req3) # 操作员完成任务，释放
        print(f"part {p_Id} OP released at {self.env.now}")
        yield self.env.timeout(0)

登录后复制