SimPy 实现简单的 DES

什么是离散事件仿真（DES）

在核心层面上，DES 是一种将系统建模为一系列特定时间点发生的离散事件的方法。每个事件都会触发系统状态的变化，比如机器的启动或停止，顾客的到来或离开，或者信号的发送或接收。通过专注于这些关键时刻，DES 提供了对系统动态的细粒度视图，而不会在连续的时间流中迷失。

DES 的关键特点

• 事件驱动动态：在 DES 中，事件之间不会发生任何事情。改变仅会在特定事件发生时发生，就像时钟的秒针仅在被推动时才会移动。
• 时间进展：时间会从一个事件跳跃到下一个事件，忽略两者之间的静默间隙。这种快进机制确保通过集中计算工作在变化时刻上，实现效率。
• 高效建模：通过仅仿真改变系统状态的事件，DES 能够高效处理复杂的现实世界系统，而无需不必要的计算开销。

DES 的应用

DES 在对时间和资源分配至关重要的行业展现出强大的优势。它的应用与它所仿真的系统一样多样化：

• 制造与生产：优化生产线、安排维护计划，并管理库存以减少停机时间并提高产量。
• 医疗保健：在医院中对患者流程进行建模，以改善等候时间、人员配置和资源利用率。
• 物流：优化供应链运营，从仓库管理到交付路线的优化，确保货物能够高效地从出发地运送到目的地。
• 电信：管理网络流量，预测拥堵点，并优化数据流以提升用户体验。
• 基础设施设计：在大型资本项目中探索“假如”情景，以在设计生命周期的早期做出明智决策，节省时间和资源。

SimPy 介绍

进入SimPy - 一个强大的、基于过程的、用于 Python 的离散事件仿真框架。 SimPy 提供了一种清晰简洁的方式来建模 DES 系统，利用了 Python 的简单性和多功能性。通过 SimPy，您可以定义事件、管理资源，并以直观高效的方式仿真过程，使其成为学术界和工业界的优秀工具。

版本历史

版本号	发布日期
2.1	2013 Oct 11
2.2	2013 Oct 11
2.3	2013 Oct 11
3.0	2013 Oct 11
3.0.3	2014 Mar 06
3.0.6	2015 Jan 30
3.0.9	2016 Jun 12
3.0.11	2018 Jul 16
3.0.13	2020 Apr 05
4.0.0	2020 Apr 07
4.1.0	2023 Nov 06
4.1.1	2023 Nov 13

SimPy 安装

想象拥有预测繁忙商店行为或无缝运营工厂的能力，而这一切都可以从您的办公桌上轻松完成。SimPy 是一个轻量级且直观的 Python 库，用于离散事件仿真，让您可以做到这一点。要开始您的仿真之旅，您只需要一个简单的命令：

$ pip install simpy

SimPy 核心概念

SimPy 围绕着一些关键概念展开，这些概念有助于仿真现实世界系统的行为。

import simpy

%load_ext jupyter_black

Processes

• 在 SimPy 中，过程使用 Python 的生成器函数来表示。这些过程会产生事件，而环境则安排这些事件的执行。
• 一个过程可以代表店铺中到达的顾客，也可以代表机器在处理物品。过程可以被暂停（使用 yield）然后在稍后恢复。

定义：

Generator = _alias(collections.abc.Generator, 3)
......
ProcessGenerator = Generator[Event, Any, Any]
......

def process(self, generator: ProcessGenerator) -> Process:
    return Process(self, generator)

Events

• 事件表示仿真中发生的特定时间点，比如机器完成任务或请求资源。
• 超时是 SimPy 中最基本的事件类型。使用 env.timeout(t)来仿真时间的流逝。其他事件包括资源请求或工序完成。

def process_example(env):
    print(f"工序开始于时间 {env.now}")
    yield env.timeout(5)
    print(f"工序结束于时间 {env.now}")

def machine(env):
    print(f"机器启动于 {env.now}")
    yield env.timeout(3)  # 机器运行3个时间单位
    print(f"机器停止于 {env.now}")

Environment

• 环境是任何 SimPy 仿真的核心。它管理仿真时钟，并控制事件的调度和执行。
• 它跟踪仿真时间，可以通过 env.now 来访问。
• 通过环境来创建和管理过程、资源和事件。

env1 = simpy.Environment()
env1.process(process_example(env1))
env1.run()

工序开始于时间 0
工序结束于时间 5

env2 = simpy.Environment()
env2.process(machine(env2))
env2.run()

机器启动于 0
机器停止于 3

env3 = simpy.Environment()
env3.process(process_example(env3))
env3.process(machine(env3))

print("仿真开始")
env3.run()
print("仿真结束")

仿真开始
工序开始于时间 0
机器启动于 0
机器停止于 3
工序结束于时间 5
仿真结束

Resources

在 SimPy 中，资源指的是模型中的共享资产，即进程竞争的服务器、机器或工作者。资源在建模具有有限可用性的系统时至关重要。

• 请求资源：当一个进程需要访问资源时，可以使用 resource.request()发出请求。
• 容量：资源可以具有有限的容量，代表可以同时为进程提供服务的实体数量（例如机器、工作者）。

使用 with resource.request() as req： 更简单、更简洁的管理资源请求的方式。

with 语句会自动处理资源的请求和释放。

示例：

resource = simpy.Resource(env, capacity=1)
with resource.request() as req:
    yield req  # 等待资源可用
    yield env.timeout(5)  # 使用资源5个时间单位

使用 resource.request() 和 resource.release() 显式管理

在更复杂的仿真中，您可能需要更多对资源请求和释放时间的控制。可以通过显式管理这些事件来实现。

def process_with_explicit_request(env, resource):
    # 请求资源
    with resource.request() as req:
        yield req  # 等待资源可用
        print(f"资源在 {env.now} 时获得")

        # 仿真使用资源
        yield env.timeout(5)
        print(f"在 {env.now} 时使用资源的过程")

        # 手动释放资源
        resource.release(req)
        print(f"资源在 {env.now} 时释放")


env = simpy.Environment()
resource = simpy.Resource(env, capacity=1)
env.process(process_with_explicit_request(env, resource))
env.run(until=10)

资源在 0 时获得
在 5 时使用资源的过程
资源在 5 时释放

解释

• 请求资源:
  • req = resource.request() 创建一个请求事件，但不立即 yield 它。这允许您精确控制进程何时应等待资源。
  • yield req 等待资源可用。
• 使用资源:
  • 在获取资源后，进程通过 yield env.timeout(5) 仿真使用它 5 个时间单位。
• 释放资源:
  • resource.release(req) 在进程使用完资源后显式释放资源。

何时使用显式资源管理

• 复杂仿真: 在更复杂的场景中，您可能需要在不同时间请求多个资源，或者资源使用取决于过程中评估的条件。
• 条件释放: 如果释放资源取决于特定条件或附加逻辑，显式管理资源使您能够灵活处理这些情况。

总结

• with resource.request() as req: 在代码块内部自动处理请求和释放，简化资源管理。
• Explicit request() and release(): 明确调用 request()和 release()函数，可以更灵活地控制资源管理的时机和条件，适用于复杂的仿真场景。

通过了解这两种方法，您可以选择最适合您仿真的复杂性和要求的方法。

SimPy 应用 - 排队系统仿真

在许多现实世界的系统中，实体（如顾客、工作或产品）需要排队等待资源变得可用（例如，服务台、机器或服务器）。SimPy 使得使用资源和进程仿真这种排队系统变得简单。

排队系统的基础知识

在排队系统中：

• 实体（例如顾客或工作）到达服务点。
• 如果服务处于忙碌状态，则实体会排队等待。
• 一旦资源可用，实体就会接受服务。
• 接受服务后，实体离开，队列中的下一个实体接受服务。

定义顾客流程

让我们仿真一个简单的系统，多个顾客到达并竞争访问单个服务台(或 SimPy 术语中的"资源")。每个顾客要么立即接受服务，要么在服务台忙碌时排队等待。

顾客流程示例：

def customer(env, name, counter):
    print(f"{name} 到达时间 {env.now}")
    with counter.request() as req:  # 请求服务台
        yield req  # 等待直到服务台可用
        print(f"{name} 正在接受服务，时间为 {env.now}")
        yield env.timeout(5)  # 仿真服务时间
        print(f"{name} 在时间 {env.now} 离开")

customer 函数仿真了顾客的行为：

• 用户到达服务柜台。
• 他们使用 counter.request()等待资源（柜台）可用。
• 一旦被服务，他们在服务台花费 5 个时间单位。
• 被服务后，他们离开系统。

定义资源（服务柜台）

在 SimPy 中，资源代表着进程（如顾客）竞争的事物。资源可以具有有限的容量，这使它们成为仿真服务台、机器或工人的理想选择。

counter = simpy.Resource(env, capacity=1)  # 只有一个服务柜台可用

添加多个顾客

现在，让我们创建多个顾客，他们将在不同时间到达并向柜台请求服务。

env = simpy.Environment()
counter = simpy.Resource(env, capacity=1)
# 创建3个在不同时间到达的顾客
env.process(customer(env, "Customer 1", counter))
env.process(customer(env, "Customer 2", counter))
env.process(customer(env, "Customer 3", counter))

env.run(until=25)  # 以15个时间单位运行仿真

Customer 1 到达时间 0
Customer 2 到达时间 0
Customer 3 到达时间 0
Customer 1 正在接受服务，时间为 0
Customer 1 在时间 5 离开
Customer 2 正在接受服务，时间为 5
Customer 2 在时间 10 离开
Customer 3 正在接受服务，时间为 10
Customer 3 在时间 15 离开

顾客到达之间的随机延迟

在现实生活中，顾客到达通常遵循某种随机过程。建模到达时间之间的常见方法是使用指数分布。我们可以通过使用 numpy 生成随机到达时间，以及 SimPy 中的 env.timeout()来处理延迟，来仿真这一过程。

import numpy as np


def customer_generator(env, counter, mean_interarrival_time):
    for i in range(5):
        env.process(customer(env, f"Customer {i+1}", counter))
        interarrival_time = np.random.exponential(mean_interarrival_time)
        yield env.timeout(interarrival_time)  # Stochastic delay between arrivals


env = simpy.Environment()
counter = simpy.Resource(env, capacity=1)

# Generate customers with stochastic delays between arrivals
mean_interarrival_time = 3  # Mean of the exponential distribution
env.process(customer_generator(env, counter, mean_interarrival_time))

env.run(until=130)

Customer 1 到达时间 0
Customer 1 正在接受服务，时间为 0
Customer 2 到达时间 1.5687061868398835
Customer 3 到达时间 2.4018411753537507
Customer 1 在时间 5 离开
Customer 2 正在接受服务，时间为 5
Customer 4 到达时间 6.347216075761495
Customer 5 到达时间 8.813850904444163
Customer 2 在时间 10 离开
Customer 3 正在接受服务，时间为 10
Customer 3 在时间 15 离开
Customer 4 正在接受服务，时间为 15
Customer 4 在时间 20 离开
Customer 5 正在接受服务，时间为 20
Customer 5 在时间 25 离开

主要区别

• 随机到达间隔时间：客户到达之间的时间现在从指数分布中绘制，使用 np.random.exponential(mean_interarrival_time)，其中 mean_interarrival_time 是到达间隔的平均延迟。
• 更逼真的客流：这仿真了更逼真的到达模式，客户以不规则间隔到达，而不是固定间隔的 3 个时间单位如原始示例中一样。

关键观察:

• 变量到达时间: 与固定间隔情况不同，顾客到达之间的时间现在是随机的。
• 队列行为: 与固定情况类似，顾客可能在另一位顾客正在接受服务时到达，并排队等待在服务柜台上轮到他们。

增加数据收集

在 process 中添加统计数据收集，以分析顾客的平均等待时间和队列长度。

# 为 statistical_data 添加 typing 注解
from typing import List, Tuple, Any

statistical_data: List[Tuple[Any]] = []


# 定义顾客进程（Customer Process）
def customer(env, name, counter):
    arrival_time = env.now
    with counter.request() as req:
        start_queue_length = len(counter.queue)
        yield req
        start_time = env.now
        yield env.timeout(5)  # 服务时间
        # print(queue_lengths)
        statistical_data.append(
            (
                arrival_time,
                start_time - arrival_time,
                start_queue_length,
                start_time,
                env.now,
                len(counter.queue),
            )
        )  # 跟踪顾客等待时间


def customer_generator(
    env, counter, mean_interarrival_time, random: bool = False, customers=10
):
    for i in range(customers):
        env.process(customer(env, f"Customer {i+1}", counter))
        if random:
            interarrival_time = np.random.exponential(mean_interarrival_time)
        else:
            interarrival_time = mean_interarrival_time

        yield env.timeout(interarrival_time)  # Stochastic delay between arrivals

statistical_data: List[Tuple[Any]] = []

env = simpy.Environment()
counter = simpy.Resource(env, capacity=1)
env.process(customer_generator(env, counter, 2))
env.run(until=100)

print(
    "到达时间",
    "等候时间",
    "当前等候人数",
    "开始服务时间",
    "结束服务时间",
    "结束是队列长度",
)
for ds in statistical_data:
    print(ds)

到达时间 等候时间 当前等候人数 开始服务时间 结束服务时间 结束是队列长度
(0, 0, 0, 0, 5, 2)
(2, 3, 1, 5, 10, 3)
(4, 6, 2, 10, 15, 5)
(6, 9, 2, 15, 20, 6)
(8, 12, 3, 20, 25, 5)
(10, 15, 3, 25, 30, 4)
(12, 18, 4, 30, 35, 3)
(14, 21, 5, 35, 40, 2)
(16, 24, 5, 40, 45, 1)
(18, 27, 6, 45, 50, 0)

可视化

import matplotlib.pyplot as plt


def plt_show():
    times = [i[0] for i in statistical_data]

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(10, 5))

    ax1.plot(times, [i[2] for i in statistical_data], marker="s")
    ax1.set_xlabel("Time")
    ax1.set_ylabel("队列长度/人数")
    ax1.grid(True)

    ax2.plot(times, [i[1] for i in statistical_data], marker="o")
    ax2.set_xlabel("Time")
    ax2.set_ylabel("等候时间/Time")
    ax2.grid(True)

    # plt.xticks(times)
    # plt.grid(True)
    plt.show()

plt_show()

• 每位顾客到达后，队列长度（包括自己）是阶梯递增的
• 每位顾客到达后的等待时间是线性增加的

进程中增加资源

class Counter(simpy.Resource):
    """simpy.Resource.capacity 是只读属性，扩展一下"""

    def __init__(self, env: simpy.Environment, capacity: int):
        super().__init__(env, capacity=capacity)

    @simpy.Resource.capacity.setter
    def capacity(self, value):
        if value <= 0:
            raise ValueError('"Capacity" must be > 0.')
        if self.count < value and self._capacity != value:
            self._capacity = value
            for _ in range(len(self.queue)):
                self._trigger_put(None)

statistical_data: List[Tuple[Any]] = []

env = simpy.Environment()
counter = Counter(env, capacity=1)
env.process(customer_generator(env, counter, 2))
env.run(30)

counter.capacity += 1
print(f"在时间 {env.now}，资源容量增加到 {counter.capacity}")
env.run()


print(
    "到达时间",
    "等候时间",
    "当前等候人数",
    "开始服务时间",
    "结束服务时间",
    "结束时队列长度",
)
for ds in statistical_data:
    print(ds)

在时间 30，资源容量增加到 2
到达时间 等候时间 当前等候人数 开始服务时间 结束服务时间 结束时队列长度
(0, 0, 0, 0, 5, 2)
(2, 3, 1, 5, 10, 3)
(4, 6, 2, 10, 15, 5)
(6, 9, 2, 15, 20, 6)
(8, 12, 3, 20, 25, 5)
(10, 15, 3, 25, 30, 3)
(12, 18, 4, 30, 35, 2)
(14, 16, 5, 30, 35, 2)
(16, 19, 5, 35, 40, 0)
(18, 17, 6, 35, 40, 0)

12、14 时刻到达的顾客，在 30 时刻同时得到服务。

plt_show()

• 每位顾客到达时的队列长度并没有变化（与不增加资源相比）—— 原因是 30 才增加的资源，队列长度统计到 18
• 每位顾客的等待时间从 14 开始出现了变化

真实场景

def real():
    global statistical_data
    statistical_data = []

    env = simpy.Environment()
    # 开始只有1个窗口
    counter = Counter(env, capacity=1)

    # 模拟20个顾客随机到达，随机函数的期望值为 3
    env.process(customer_generator(env, counter, 3, random=True, customers=20))
    env.run(30)
    counter.capacity += 1
    print(f"在时间 {env.now}，资源容量增加到 {counter.capacity}")
    env.run()

    plt_show()

real()

在时间 30，资源容量增加到 2

• 顾客到来而相对平均，尤其中间有一个长期的不来人，让后面到来的都享受到红利
• 甚至40+ 到来的顾客可以不排队就直接得到服务

real()

在时间 30，资源容量增加到 2

• 前期来人太快，导致队伍一直很长，降不下来
• 从第6个人开始，等待的时间一直高居不下

real()

在时间 30，资源容量增加到 2

• 非常幸运的过程，前期 counter 少的时候来人慢
• 后期 couter 增加了，来人也没有爆发

更多贴近真实

• 顾客到来是阶段性随机的，即：每个时间段内随机函数的数学期望不一样。
• counter 的增加与顾客队伍的长度形成一个正相关关系，即：管理者根据队伍的长度决定增加 counter 的数量。
• coutner 也有休息的时间。

高级话题

• 多 process 并发
• 状态机：SimPy 可以与状态机库结合使用，实现复杂的模拟逻辑。
• 事件驱动（Event-Driven Concurrency）
• Resource
  • 有优先级的 Resource
  • 可抢占的 Resource
• SimPy 仿真应用：
  • 网络场景：路由、流量控制……
  • 生产制造：工序管理和布局……
  • 交通系统：交通流量、道路阻塞……

Process 并发

import simpy


def p1(env):
    while True:
        yield env.timeout(2)
        print(f"{env.now}: 任务1 执行完毕")


def p2(env, res):
    with res.request() as req:
        yield req
        print(f"{env.now}: 任务2 得到资源")
        yield env.timeout(3)
        print(f"{env.now}: 任务2 执行完毕")


env = simpy.Environment()
env.process(p1(env))
env.process(p2(env, simpy.Resource(env, capacity=1)))
env.run(until=10)

0: 任务2 得到资源
2: 任务1 执行完毕
3: 任务2 执行完毕
4: 任务1 执行完毕
6: 任务1 执行完毕
8: 任务1 执行完毕

状态机

import simpy


def state_machine(env):
    states = ["A", "B", "C"]
    current_state = states[0]

    while True:
        print(f"{env.now}: 当前状态 {current_state}")
        if current_state == "A":
            yield env.timeout(1)
            current_state = "B"
        elif current_state == "B":
            yield env.timeout(2)
            current_state = "C"
        else:
            yield env.timeout(3)
            current_state = "A"


env = simpy.Environment()
env.process(state_machine(env))
env.run(until=10)

0: 当前状态 A
1: 当前状态 B
3: 当前状态 C
6: 当前状态 A
7: 当前状态 B
9: 当前状态 C

事件驱动

import simpy


def task(env, event):
    print(f"{env.now}: 任务2 等待事件")
    yield event
    print(f"{env.now}: 任务2 执行")


def trigger_event(env, event):
    yield env.timeout(5)  # 在5个时间单位后触发事件
    print(f"{env.now}: 触发事件")
    event.succeed()


# 创建环境和事件
env = simpy.Environment()
event = simpy.Event(env)

# 启动任务1和任务2进程
env.process(task(env, event))

# 启动事件触发器进程
env.process(trigger_event(env, event))

# 运行模拟，直到10个时间单位
env.run(until=10)

0: 任务2 等待事件
5: 触发事件
5: 任务2 执行