psim言語講座（第15回）チュートリアル編（6）（続）psim のストアを使いこなす

HOME
psim言語講座（第15回）チュートリアル編（6）（続）psim のストアを使いこなす

2024年7月10日 09:38

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

はじめに

前回は、Store の応用例として、その背後にある psim のスケジューラを説明しました。

今回は、もう少し Store の応用例を説明していきます。

Pythonオブジェクトを利用した廃棄処理

前回、ストアを利用した廃棄処理の例を示しました。その例では、Python の辞書をストアに格納していましたが、今回は、Python のオブジェクトをストアに格納する方法を説明します。

シミュレーション対象としては、スーパーの商品棚の例で、商品の供給と消費が自立的に発生する中、毎時間棚卸しを行う例を考えます。棚卸しは、消費期限切れの商品を廃棄します。

まず商品クラスを定義します。

    class Item:
        def __init__(self, name, disposalTime):
            self.name = name
            self.disposalTime = disposalTime

特に難しい事はなく、商品は、名前 name と、消費時間 disposalTime を持つとしています。

前回の例では辞書をストアに加えていたので、以下のようなコードがありました。

    yield s.put1({"商品": "商品名" "廃棄時間": now() + 有効期間})

今回は製品クラスを利用して、より自然に以下のように書く事が出来ます。

    item = Item("商品名", now() + 有効期間)
    yield s.put1(item)

前回の例では商品品取得時の条件式は、以下のように書いていました。

        result = yield s.get1(lambda v: v["商品"] == item, name = "get1") | pause(0, name = "giveup")

製品クラスを利用した場合は、以下のように書く事が出来ます。

        result = yield s.get1(lambda v: v.name == itemname, name = "get1") | pause(0, name = "giveup")

全体のコードは以下のように記述できます。

    initialize()
    s = Store(False, monitor = True)

    class Item:
        def __init__(self, name, disposalTime):
            self.name = name
            self.disposalTime = disposalTime

    def supply(item):
        print(f"時間 {now()} {item.name} 追加待ち")
        printStoreStatus()
        yield s.put1(item)
        print(f"時間 {now()} {item.name} 追加終了")
        printStoreStatus()

    def supplier():
        for itemname, validityPeriod in [("商品1", 3),
                                         ("商品2", 1),
                                         ("商品3", 2),
                                         ("商品4", 1),
                                         ("商品5", 2),
                                         ("商品6", 3),
                                         ("商品7", 2)]:
            yield pause(1)
            yield subactivate(supply)(Item(itemname, now() + validityPeriod))
        yield alwaysFalse()

    def consume(itemname):
        print(f"時間 {now()} {itemname} 取得待ち")
        printStoreStatus()
        result = yield s.get1(lambda v: v.name == itemname, name = "get1") | pause(0, name = "giveup")
        if "giveup" in result:
            print(f"時間 {now()} {itemname} 取得断念")
            printStoreStatus()
        else:
            print(f"時間 {now()} {itemname} 取得終了")
            printStoreStatus()

    def consumer():
        for itemname in ["商品4", "商品1", "商品2", "商品7", "商品5",
                         "商品6", "商品3"]:
            yield pause(2)
            yield subactivate(consume)(itemname)
        yield alwaysFalse()

    def disposer():
        while True:
            yield pause(1)
            while True:
                result = yield s.get1(lambda v: v.disposalTime >= now(),
                                     name = "get1") | pause(0, name = "giveup")
                if "giveup" in result:
                    break
                else:
                    itemname = result["get1"].name
                    print(f"時間 {now()} {itemname} 廃棄")
            printStoreStatus()

    activate(supplier)()
    activate(consumer)()
    activate(disposer)()
    start(until = 10)

実行結果は以下のようになります。

バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 1.0 商品1 追加待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 1.0 商品1 追加終了
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品1 廃棄
時間 2.0 商品4 取得待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品4 取得断念
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加終了
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 3.0 商品2 廃棄

ストアから納期の最短のものを取得

次に納期最短のものを取得する例を考えてみます。廃棄処理とはちょっとだけ処理が違います。廃棄処理では、廃棄対象の商品がみつかると廃棄していました。一方で、今回は、全商品の中から納期の最短のものを探し出し、それを取得する必要があります。そのためには、全商品をリストアップし、納期が最短のものを探し、その製品を取得する必要があります。その操作を実現するには、製品を一意に示す製品IDという概念が必要になります。

今回は、ストアの中には辞書を持つ方式で実装してみます。

例えば、以下のように、製品をストアに格納します。

        yield s.put1({"商品": "商品1",
                      "納期": 123,
                      "製品ID": 456})

ここで、製品IDはユニークなIDになるようにします。通常はカウンターを用意して、追加するごとに、そのカウンターをインクリメントする方法が簡単です。

実際に、納期最小順の製品リストを取得するには、以下のようなコードになります。

        # 納期最小順の製品リスト
        lst = sorted(s.buffer, key = lambda e: e["納期"])
        if len(lst) > 0:
            productid = lst[0]["製品ID"] # 納期最小の製品
            deadline = lst[0]["納期"]
            result = yield s.get1(lambda v: v["製品ID"] == productid, name = "get1") | pause(0, name = "giveup")

全体のコードは以下のようになります。

    initialize()
    s = Store(False, monitor = True)

    def supply(item, deadline, productid):
        print(f"時間 {now()} {item} 追加待ち")
        printStoreStatus()
        yield s.put1({"商品": item, "納期": deadline, "製品ID": productid})
        print(f"時間 {now()} {item} 追加終了")
        printStoreStatus()

    def supplier():
        productid = 0
        for item, deadline in [("商品1", 3),
                               ("商品2", 1),
                               ("商品3", 2),
                               ("商品4", 1),
                               ("商品5", 2),
                               ("商品6", 3),
                               ("商品7", 2)]:
            yield pause(0)
            yield subactivate(supply)(item, deadline, productid)
            productid += 1
        yield alwaysFalse()

    def consume():
        print(f"時間 {now()} 取得待ち")
        printStoreStatus()
        # 納期最小順の製品リスト
        lst = sorted(s.buffer, key = lambda e: e["納期"])
        if len(lst) > 0:
            productid = lst[0]["製品ID"] # 納期最小の製品
            deadline = lst[0]["納期"]
            result = yield s.get1(lambda v: v["製品ID"] == productid, name = "get1") | pause(0, name = "giveup")
            if "giveup" in result:
                print(f"時間 {now()} 製品 {productid} 納期 {deadline} 取得断念")
                printStoreStatus()
            else:
                print(f"時間 {now()} 製品 {productid} 納期 {deadline} 取得終了")
                printStoreStatus()

    def consumer():
        for i in range(7):
            yield pause(2)
            yield subactivate(consume)()
        yield alwaysFalse()

    activate(supplier)()
    activate(consumer)()
    start(until = 30)

ストアとファシリティの組み合わせ

次に、より現実的な、工場の生産ラインの一部を再現してみます。

工場の生産ラインには、加工機械が複数あります。そのひとつの加工機械のみに着目します。

加工機械の前には製品の置き場(s)があります。加工機械(f)は、製品の置き場から製品をひとつ取得し、加工し、加工済みの製品を、加工機械の後の製品置き場(t)におきます。

それぞれ、以下にように定義します。

    s = Store(False, monitor = True)
    f = Facility(1, monitor = True)
    t = Store(False, monitor = True)

各機械の動作は、以下のように書けます。

        while True:
            result = yield s.get1(name = "get")
            item = result["get"]
            print(f"時間 {now()} 作業者 {worker} 作業開始 {item}")
            lock = (yield f.request(name = "lock"))["lock"]
            yield pause(10)
            lock.release()
            print(f"時間 {now()} 作業者 {worker} 作業終了 {item}")
            yield t.put1(item, name = "put")

全体のコードは以下のようになります。

    initialize()
    s = Store(False, monitor = True)
    f = Facility(1, monitor = True)
    t = Store(False, monitor = True)

    def init():
        for i in range(10):
            yield s.put1(f"製品{i + 1}")
    def process(worker):
        yield pause(10)
        while True:
            result = yield s.get1(name = "get")
            item = result["get"]
            print(f"時間 {now()} 作業者 {worker} 作業開始 {item}")
            lock = (yield f.request(name = "lock"))["lock"]
            yield pause(10)
            lock.release()
            print(f"時間 {now()} 作業者 {worker} 作業終了 {item}")
            yield t.put1(item, name = "put")
    activate(init)()
    for i in range(3):
        activate(process)(f"作業者{i + 1}")
    start()

    fig, axes = plt.subplots(1, 2)

    time = s.monitor.time().toList()
    buf = s.monitor["バッファ"].toList()
    axes[0].step(time, buf, where = "post")
    axes[0].set_title("ストア s")
    axes[0].set_xlabel("時間")
    axes[0].set_ylabel("在庫")

    time = t.monitor.time().toList()
    buf = t.monitor["バッファ"].toList()
    axes[1].step(time, buf, where = "post")
    axes[1].set_title("ストア t")
    axes[1].set_xlabel("時間")
    axes[1].set_ylabel("在庫")

    plt.tight_layout()
    plt.show()

それぞれのストアのバッファ数の時系列変化をプロットすると以下のようになります。