【Python】Simpyではじめる離散イベントシミュレーション -作業工程を見直して生産性向上のヒントを得る-

こんにちは。現役エンジニアの”はやぶさ”@Cpp_Learningです。社内のDX推進を頑張ってます！

本記事では離散イベントシミュレーションフレームワークのSimpyを活用した、作業工程の見直しについて説明します。

Contents

1 離散イベントシミュレーションを行うモチベーション
2 実践！Simpyではじめる離散イベントシミュレーション
3 まとめ -Simpyではじめる離散イベントシミュレーション-

離散イベントシミュレーションを行うモチベーション

複数の工程が必要な作業や実験をする際、制限時間内で何サイクル回せるかを事前確認したいときがよくあります。

実験サイクル例

準備：5[sec]
工程A：10[sec]
工程B：8[sec]
工程C：12[sec]
片付け：10[sec]

上の例では、1サイクル回すのに45[sec]かかるので、100[sec]では2サイクル+工程Aの途中で終了します。これくらいの時間の見積もりなら、それほど難しくありませんが、各工程でばらつきがある場合はどうでしょうか？

実験サイクル例 -ばらつき有-

準備：5[sec]
工程A：10[sec]
工程B：8 + σ [sec]、σ = 0～10[sec]
工程C：12 + σ [sec]、σ = 0～10[sec]
片付け：10[sec]

例えば、工程Bと工程Cでσ=0～10[sec] のばらつきが発生すると仮定したとき、100[sec]以内に何サイクル回せるかをシミュレーションで確認したくなります。

さらに作業員が2人に増えたときや工程Aで待ちが発生するなどの複数の制約も考慮したシミュレーションをしたいことがあります。

ここまで説明したシミュレーションについては Simpy を使えば、比較的簡単に実現できます。

実践！Simpyではじめる離散イベントシミュレーション

簡単なシミュレーションから始め、少しづつ複雑なシミュレーションを実践していきます。

インストール

最初に以下のコマンドで Simpy をインストールします。

pip install simpy

以降からソースコード書いていきます。

本記事のソースコードはGoogle Colabで動作確認しました

Simpyの基本的な使い方

上図の実験サイクルは、以下のコードでシミュレーションできます。

import simpy 


def experiment(env):
    """ 実験手順 """
    n = 1
    while True: # Simulate until the time limit
        print(f'==== Start experiment No.{n} at {env.now} [sec]  ==== ')
        running_time = 5
        yield env.timeout(running_time) # models duration

        print(f'Process A at {env.now} [sec]')
        running_time = 10
        yield env.timeout(running_time)

        print(f'Process B at {env.now} [sec]')
        running_time = 12
        yield env.timeout(running_time)

        print(f'Process C at {env.now} [sec]')
        running_time = 8
        yield env.timeout(running_time)

        print(f'==== Finish experiment No.{n} at {env.now} [sec] ==== ')
        running_time = 10
        yield env.timeout(running_time)
        n = n + 1
        
# Create environment
env = simpy.Environment() 
env.process(experiment(env))

# Run simulation until 100 [sec]
env.run(until=100)

import simpy

def experiment(env):

""" 実験手順 """

n = 1

while True: # Simulate until the time limit

print(f'==== Start experiment No.{n} at {env.now} [sec] ==== ')

running_time = 5

yield env.timeout(running_time) # models duration

print(f'Process A at {env.now} [sec]')

running_time = 10

yield env.timeout(running_time)

print(f'Process B at {env.now} [sec]')

running_time = 12

yield env.timeout(running_time)

print(f'Process C at {env.now} [sec]')

running_time = 8

yield env.timeout(running_time)

print(f'==== Finish experiment No.{n} at {env.now} [sec] ==== ')

running_time = 10

yield env.timeout(running_time)

n = n + 1

# Create environment

env = simpy.Environment()

env.process(experiment(env))

# Run simulation until 100 [sec]

env.run(until=100)

==== Start experiment No.1 at 0 [sec] ====
Process A at 5 [sec]
Process B at 15 [sec]
Process C at 27 [sec]
==== Finish experiment No.1 at 35 [sec] ====
==== Start experiment No.2 at 45 [sec] ====
Process A at 50 [sec]
Process B at 60 [sec]
Process C at 72 [sec]
==== Finish experiment No.2 at 80 [sec] ====
==== Start experiment No.3 at 90 [sec] ====
Process A at 95 [sec]

上記がターミナルに表示され、100[sec]では3サイクル目の工程Aの途中で終了することを確認できます。

このコードを雛形に少しづつ条件や制約を増やしたシミュレーションをしていきます。

Simpyでばらつきを考慮したシミュレーション

工程Bと工程Cでσ=0～10[sec]のばらつき発生を仮定したシミュレーションは random 追加で実現できます。

import simpy 
from random import random, seed

# seed(1)


def experiment(env):
    """ 実験手順 """
    n = 1
    while True: # Simulate until the time limit
        print(f'==== Start experiment No.{n} at {round(env.now, 2)} [sec]  ==== ')
        running_time = 5
        yield env.timeout(running_time) # models duration

        print(f'Process A at {round(env.now, 2)} [sec]')
        running_time = 10
        yield env.timeout(running_time)

        print(f'Process B at {round(env.now, 2)} [sec]')
        # random time is between 0 and 10 sec
        running_time = 12 + random() * 10
        yield env.timeout(running_time)

        print(f'Process C at {round(env.now, 2)} [sec]')
        # random time is between 0 and 10 sec
        running_time = 8 + random() * 10
        yield env.timeout(running_time)

        print(f'==== Finish experiment No.{n} at {round(env.now, 2)} [sec] ==== ')
        running_time = 10
        yield env.timeout(running_time)
        n = n + 1

# Create environment
env = simpy.Environment() 
env.process(experiment(env))

# Run simulation until 100 [sec]
env.run(until=100)

import simpy

from random import random, seed

# seed(1)

def experiment(env):

""" 実験手順 """

n = 1

while True: # Simulate until the time limit

print(f'==== Start experiment No.{n} at {round(env.now, 2)} [sec] ==== ')

running_time = 5

yield env.timeout(running_time) # models duration

print(f'Process A at {round(env.now, 2)} [sec]')

running_time = 10

yield env.timeout(running_time)

print(f'Process B at {round(env.now, 2)} [sec]')

# random time is between 0 and 10 sec

running_time = 12 + random() * 10

yield env.timeout(running_time)

print(f'Process C at {round(env.now, 2)} [sec]')

# random time is between 0 and 10 sec

running_time = 8 + random() * 10

yield env.timeout(running_time)

print(f'==== Finish experiment No.{n} at {round(env.now, 2)} [sec] ==== ')

running_time = 10

yield env.timeout(running_time)

n = n + 1

# Create environment

env = simpy.Environment()

env.process(experiment(env))

# Run simulation until 100 [sec]

env.run(until=100)

==== Start experiment No.1 at 0 [sec] ====
Process A at 5 [sec]
Process B at 15 [sec]
Process C at 30.4 [sec]
==== Finish experiment No.1 at 41.53 [sec] ====
==== Start experiment No.2 at 51.53 [sec] ====
Process A at 56.53 [sec]
Process B at 66.53 [sec]
Process C at 81.44 [sec]
==== Finish experiment No.2 at 99.1 [sec] ====

シード固定しなければ、毎回ランダムなばらつきが発生します。上記では 99.1[sec]で2サイクル目が終了していますが、2サイクル目の工程Cの途中で終了するケースもありました。

つまり100[sec]以内で2サイクルがギリギリ終わるときと、終わらないときがあるというシミュレーション結果が得られました。

作業員を増やしてシミュレーション

作業員を2人に増やしてみます。以下のように、operator関連のコードを追加すればOKです。

import simpy 
from random import random, seed

# seed(1)


def experiment(env, operator):
    """ 実験手順 """
    n = 1
    while True: # Simulate until the time limit
        print(f'==== {operator} : Start experiment No.{n} at {round(env.now, 2)} [sec]  ==== ')
        running_time = 5
        yield env.timeout(running_time) # models duration

        print(f'{operator} : Process A at {round(env.now, 2)} [sec]')
        running_time = 10
        yield env.timeout(running_time)

        print(f'{operator} : Process B at {round(env.now, 2)} [sec]')
        # random time is between 0 and 10 sec
        running_time = 12 + random() * 10
        yield env.timeout(running_time)

        print(f'{operator} : Process C at {round(env.now, 2)} [sec]')
        # random time is between 0 and 10 sec
        running_time = 8 + random() * 10
        yield env.timeout(running_time)

        print(f'==== {operator} : Finish experiment No.{n} at {round(env.now, 2)} [sec] ==== ')
        running_time = 10
        yield env.timeout(running_time)
        n = n + 1

# Create environment
env = simpy.Environment() 
# env.process(experiment(env))

operator_num = 2
for i in range(operator_num):
    env.process(experiment(env, f'Operator No.{i}'))

# Run simulation until 100 [sec]
env.run(until=100)

import simpy

from random import random, seed

# seed(1)

def experiment(env, operator):

""" 実験手順 """

n = 1

while True: # Simulate until the time limit

print(f'==== {operator} : Start experiment No.{n} at {round(env.now, 2)} [sec] ==== ')

running_time = 5

yield env.timeout(running_time) # models duration

print(f'{operator} : Process A at {round(env.now, 2)} [sec]')

running_time = 10

yield env.timeout(running_time)

print(f'{operator} : Process B at {round(env.now, 2)} [sec]')

# random time is between 0 and 10 sec

running_time = 12 + random() * 10

yield env.timeout(running_time)

print(f'{operator} : Process C at {round(env.now, 2)} [sec]')

# random time is between 0 and 10 sec

running_time = 8 + random() * 10

yield env.timeout(running_time)

print(f'==== {operator} : Finish experiment No.{n} at {round(env.now, 2)} [sec] ==== ')

running_time = 10

yield env.timeout(running_time)

n = n + 1

# Create environment

env = simpy.Environment()

# env.process(experiment(env))

operator_num = 2

for i in range(operator_num):

env.process(experiment(env, f'Operator No.{i}'))

# Run simulation until 100 [sec]

env.run(until=100)

==== Operator No.0 : Start experiment No.1 at 0 [sec] ====
==== Operator No.1 : Start experiment No.1 at 0 [sec] ====
Operator No.0 : Process A at 5 [sec]
Operator No.1 : Process A at 5 [sec]
Operator No.0 : Process B at 15 [sec]
Operator No.1 : Process B at 15 [sec]
Operator No.1 : Process C at 31.95 [sec]
Operator No.0 : Process C at 34.24 [sec]
==== Operator No.1 : Finish experiment No.1 at 40.53 [sec] ====
==== Operator No.0 : Finish experiment No.1 at 45.21 [sec] ====
==== Operator No.1 : Start experiment No.2 at 50.53 [sec] ====
==== Operator No.0 : Start experiment No.2 at 55.21 [sec] ====
Operator No.1 : Process A at 55.53 [sec]
Operator No.0 : Process A at 60.21 [sec]
Operator No.1 : Process B at 65.53 [sec]
Operator No.0 : Process B at 70.21 [sec]
Operator No.1 : Process C at 86.97 [sec]
Operator No.0 : Process C at 88.34 [sec]
==== Operator No.1 : Finish experiment No.2 at 98.76 [sec] ====

Operator No.0 と No.1 は同時に作業を開始しますが、工程BとCでばらつきが発生するため、以下のシミュレーション結果となりました。

Operator No.0：2サイクル目の工程Cの途中で100[sec]経過
Operator No.1：98.76[sec]で2サイクル目まで終了

待ちが発生するシミュレーション

リソースの都合で、並行処理できないケースもあります。例えば、上図の工程Aは専用機械を使うため、1回の実行で1つの処理しかできないと仮定します。つまり、Operator No.0 が工程Aを終了するまで、Operator No.1 は待機する必要があります。

このようなリソース制約は simpy.resource で定義できます。

import simpy 
from random import random, seed

# seed(1)


def experiment(env, operator, machine_a):
    """ 実験手順 """
    n = 1
    while True: # Simulate until the time limit
        print(f'==== {operator} : Start experiment No.{n} at {round(env.now, 2)} [sec]  ==== ')
        running_time = 5
        yield env.timeout(running_time) # models duration

        with machine_a.request() as m_req:
            yield m_req 

            print(f'{operator} : Process A at {round(env.now, 2)} [sec]')
            running_time = 10
            yield env.timeout(running_time)

        print(f'{operator} : Process B at {round(env.now, 2)} [sec]')
        # random time is between 0 and 10 sec
        running_time = 12 + random() * 10
        yield env.timeout(running_time)

        print(f'{operator} : Process C at {round(env.now, 2)} [sec]')
         # random time is between 0 and 10 sec
        running_time = 8 + random() * 10
        yield env.timeout(running_time)

        print(f'==== {operator} : Finish experiment No.{n} at {round(env.now, 2)} [sec] ==== ')
        running_time = 10
        yield env.timeout(running_time)
        n = n + 1

# Create environment
env = simpy.Environment() 

# Machine A that can only allow 1 operator at once
machine_a = simpy.Resource(env, capacity=1)

operator_num = 2
for i in range(operator_num):
    env.process(experiment(env, f'Operator No.{i}', machine_a))

# Run simulation until 100 [sec]
env.run(until=100)

import simpy

from random import random, seed

# seed(1)

def experiment(env, operator, machine_a):

""" 実験手順 """

n = 1

while True: # Simulate until the time limit

print(f'==== {operator} : Start experiment No.{n} at {round(env.now, 2)} [sec] ==== ')

running_time = 5

yield env.timeout(running_time) # models duration

with machine_a.request() as m_req:

yield m_req

print(f'{operator} : Process A at {round(env.now, 2)} [sec]')

running_time = 10

yield env.timeout(running_time)

print(f'{operator} : Process B at {round(env.now, 2)} [sec]')

# random time is between 0 and 10 sec

running_time = 12 + random() * 10

yield env.timeout(running_time)

print(f'{operator} : Process C at {round(env.now, 2)} [sec]')

# random time is between 0 and 10 sec

running_time = 8 + random() * 10

yield env.timeout(running_time)

print(f'==== {operator} : Finish experiment No.{n} at {round(env.now, 2)} [sec] ==== ')

running_time = 10

yield env.timeout(running_time)

n = n + 1

# Create environment

env = simpy.Environment()

# Machine A that can only allow 1 operator at once

machine_a = simpy.Resource(env, capacity=1)

operator_num = 2

for i in range(operator_num):

env.process(experiment(env, f'Operator No.{i}', machine_a))

# Run simulation until 100 [sec]

env.run(until=100)

==== Operator No.0 : Start experiment No.1 at 0 [sec] ====
==== Operator No.1 : Start experiment No.1 at 0 [sec] ====
Operator No.0 : Process A at 5 [sec]
Operator No.0 : Process B at 15 [sec]
Operator No.1 : Process A at 15 [sec]
Operator No.1 : Process B at 25 [sec]
Operator No.0 : Process C at 27.51 [sec]
==== Operator No.0 : Finish experiment No.1 at 41.45 [sec] ====
Operator No.1 : Process C at 44.53 [sec]
==== Operator No.0 : Start experiment No.2 at 51.45 [sec] ====
Operator No.0 : Process A at 56.45 [sec]
==== Operator No.1 : Finish experiment No.1 at 57.11 [sec] ====
Operator No.0 : Process B at 66.45 [sec]
==== Operator No.1 : Start experiment No.2 at 67.11 [sec] ====
Operator No.1 : Process A at 72.11 [sec]
Operator No.0 : Process C at 80.0 [sec]
Operator No.1 : Process B at 82.11 [sec]
==== Operator No.0 : Finish experiment No.2 at 92.51 [sec] ====