psim言語講座（第4回）生産ラインシミュレーションを書いてみる

HOME
psim言語講座（第4回）生産ラインシミュレーションを書いてみる

2021年10月15日 15:31

本記事は当社が発行しているシミュレーションメールマガジンVol.4の記事です。
シミュレーションメールマガジンの詳細・購読申込はこちらのサポートページから

生産ラインモデル

前回まで M/M/1 モデルを解説してきました。

M/M/1 モデルとは、同時に 1 つのみ行えるサービスに対し、ランダムに利用者が到着し、ランダムな時間サービスを利用するようなモデルの事でした。

今回はそれを拡張して何らかの製品を生産するような生産ラインモデルを考えていきます。

以下のような工程が 5 つあるような生産ラインを考えます。最初に op1 にて、製品を加工を、順番に、op2, op3, o4, op5 と加工を進めていきます。

グラフ

工程ごとの、並列数、入力バッファ数、処理時間などは以下のように定義されています。

工程名	入力バッファ数	並列数	処理時間平均	処理時間標準偏差
op1	0	1	9	2
op2	0	2	29	2
op3	3	1	9	2
op4	0	3	25	2
op5	0	1	9	2

並列数というのはその工程には複数の加工機械がある事を想定しており、加工機械分、並列に加工が行えます。

入力前バッファというのは、加工機械の前に製品の置き場が用意されていて、指定した個数の製品を蓄えておく事ができます。

処理時間平均、処理時間標準偏差というのは、加工に要する時間(正規分布)を示します。

製品クラス

まず製品クラスを作成します。

class Item:
    def __init__(self, name):
        self.name = name
        self.dic = {}
    def recordEnter(self, mname, time):
        self.dic[mname] = [time, time]
    def recordExit(self, mname, time):
        self.dic[mname][1] = time
    def getLog(self):
        return [(m, self.name, t0, t1) for m, (t0, t1) in self.dic.items()]
    def __repr__(self):
        return self.name

このクラスは、簡単な Python のクラスで、特に複雑な事はしていませんが、後で、各加工機械への入力時間、出力時間を集計しやすいようなメソッドを用意しています。

加工機械クラス

次が加工機械クラスです。

class Machine:
    def __init__(self, name, nbuf, npar, mu, sig):
        self.name = name
        self.nbuf = nbuf
        self.npar = npar
        self.mu = mu
        self.sig = sig
        if self.nbuf == 0:
            self.inputQueue = Store("empty")
        else:
            self.inputQueue = Store(self.nbuf)
        for i in range(self.npar):
            activate(self.proc)(i)
        self.next = None
    def proc(self, i):
        def machinename(i):
            if self.npar == 1:
                return self.name
            else:
                return "%s-%d" % (self.name, i + 1)
        while True:
            result = yield self.inputQueue.get1(name = "item")
            item = result["item"]
            item.recordEnter(machinename(i), now())
            yield pause(np.random.normal(self.mu, self.sig))
            item.recordExit(machinename(i), now())
            if self.next is not None:
                yield self.next.put(item)
    def setNext(self, machine):
        self.next = machine
        return self.name
    def put(self, item):
        def _():
            yield self.inputQueue.put1(item)
        return call(_)()

proc メソッドが、加工処理プロセスを実装しています。内部では、M/M/1 モデルで紹介した、Facility と pause を使用しています。また、後でガントチャートを出力できるようにするため、加工機械を通過する各製品オブジェクトに、加工機械への入力時間、出力時間を記録しています。

なお、加工プロセスの入り口は入力バッファを経由するように実装しています。この入力バッファは psim の機能である Store を利用して表現しています。

            self.inputQueue = Store(self.nbuf)

は、容量 self.nbuf の入力バッファを作成しています。("empty" は容量のない特別な Store を作成します)

            yield self.inputQueue.put1(item)

は、入力バッファに製品を追加するのを待ち受けることを表現しています。

            result = yield self.inputQueue.get1(name = "item")
            item = result["item"]

は、入力バッファから製品を取り出しを待ち受けることを表現しています。

加工プロセス

以上の部品を組み合わせて、生産ラインシミュレーションを作成します。

initialize()

# 加工機械を作成
machines = []    
for i, args in enumerate(mlist):
    m = Machine(*args)
    machines.append(m)

# 加工機械を結合
for i in range(len(machines) - 1):
    machines[i].setNext(machines[i + 1])

items = []

def put(item):
    yield machines[0].put(item)

# 生産ラインに製品を投入するプロセス
def proc():
    for i in range(nitems):
        item = Item("item%2d" % i)
        items.append(item)
        activate(put)(item)
    yield alwaysTrue()

# シミュレーション開始
activate(proc)()        
start(until = None)

ガントチャート作成

def plot_gantt(ax, sched, title):
    machines = set([machine for machine, job, start, end in sched])
    machines = sorted(list(machines))
    machines = {m: i for i, m in enumerate(machines)}
    items = set([job for machine, job, start, end in sched])
    items = sorted(list(items))
    items = {m: i for i, m in enumerate(items)}
    starttimes = [start for machine, job, start, end in sched]
    endtimes = [end for machine, job, start, end in sched]
    colors = matplotlib.cm.tab20.colors
    for machine, job, start, end in sched:
        ax.broken_barh([(start, end - start)], (machines[machine] - 0.4, 0.8),
                       color = colors[items[job] % len(colors)])
    ax.set_yticks(range(len(machines)))
    ax.set_yticklabels(machines)
    ax.set_xlabel("time")
    ax.set_xlim((min(starttimes), max(endtimes)))
    ax.invert_yaxis()
    ax.set_title(title)

sched = []
for item in items:
    sched.extend(item.getLog())

endtimes = [end for machine, job, start, end in sched]
print(max(endtimes)) # 終了時間

fig, axes = plt.subplots(1, 1)
plot_gantt(axes, sched, "gantt chart")
plt.show()

ガントチャートを表示します。