simpy

SimPy - 離散イベントシミュレーション

SimPy では、Python のジェネレーターとして現実世界のプロセスをモデル化できます。SimPy のコンポーネント（顧客、車、データパケットなど）は、互いに、また限られた「リソース」（サーバー、駐車スペース、帯域幅など）とやり取りする「プロセス」です。

使用する場面

• 待ち行列システムのモデル化（銀行の窓口、コールセンター、病院の救急外来）
• サプライチェーンと物流のシミュレーション（倉庫、輸送ネットワーク）
• 製造プロセスの分析（組立ライン、機械のメンテナンス）
• ネットワークシミュレーション（パケットルーティング、サーバー負荷分散）
• プロジェクト管理（タスクの依存関係、リソースの割り当て）
• イベントが連続的ではなく、特定の時点で発生するあらゆるシステム

参考ドキュメント

公式ドキュメント：https://simpy.readthedocs.io/
GitHub：https://github.com/simpy/simpy
検索パターン：simpy.Environment、simpy.Resource、env.process、yield env.timeout

基本原則

仮想時間

SimPy は 0 から始まる内部時計を使用します。イベントが処理されたときにのみ、時間が進みます。イベント間では何も起きないため、イベントが少ししか発生しない場合、100 年分のシミュレーションも数秒で完了します。

ジェネレーターとしてのプロセス

プロセスは yield を使う通常の Python 関数です。プロセスがイベントを yield すると、SimPy はそのイベントが発生するまでプロセスを一時停止します。

リソース

共有オブジェクトには次の 3 種類があります。

• Resource：数に限りのあるスロット（カウンターなど）
• Container：まとまった量の物質（ガスタンク、RAM など）
• Store：個別のオブジェクト（メッセージのバッファーなど）

クイックリファレンス

インストール

pip install simpy

標準インポート

import simpy
import random

基本パターン - シンプルなプロセス

import simpy

def clock(env, name, tick):
    while True:
        print(f"{name} at {env.now}")
        # 'yield' tells SimPy to wait for this event
        yield env.timeout(tick)

# 1. Create Environment
env = simpy.Environment()

# 2. Add Process
env.process(clock(env, 'Fast', 0.5))
env.process(clock(env, 'Slow', 1.0))

# 3. Run for 2 units of virtual time
env.run(until=2.1)

重要なルール

✅ 推奨事項

• yield env.timeout(n) を使用する - プロセスの時間を進める方法はこれだけです。time.sleep() は使用しないでください。
• with resource.request() as req: を使用する - プロセスが中断または失敗した場合も、リソースが自動的に解放されます。
• 乱数シードを設定する - シミュレーションを再現可能にするため、必ず random.seed(42) を使用してください。
• env オブジェクトを渡す - 時間を追跡するため、すべてのプロセスとリソースに環境への参照が必要です。
• env.run(until=...) を使用する - 必ず停止条件を定義してください。定義しないと、while True のシミュレーションは永久に実行されます。
• データをリストに収集する - シミュレーション後の分析（Pandas の利用など）に備えて、タイムスタンプとイベントの詳細を外部リストに保存してください。

❌ 禁止事項

• 現実世界の時間を使用する - simpy は仮想時間用です。時計として使うのは env.now のみにしてください。
• 環境を直接変更する - 環境とは、プロセスとイベントを通じてのみやり取りしてください。
• yield を忘れる - ループ内でイベントを yield しないと、仮想時間の同じ瞬間に無限ループが発生し、シミュレーションが停止します。
• イベントループをブロックする - プロセス内での負荷の高い CPU 計算は、できるだけ避けてください。CPU が計算している間、シミュレーション全体の仮想時計が「止まります」。

アンチパターン（絶対に避ける）

import simpy
import time

# ❌ BAD: Using real sleep
def bad_process(env):
    time.sleep(1) # This stops the REAL clock, but virtual time doesn't move!
    yield env.timeout(1)

# ✅ GOOD: Use virtual timeout
def good_process(env):
    yield env.timeout(1) # Moves virtual clock forward instantly

# ❌ BAD: Infinite loop without yielding
# def hang_forever(env):
#     while True:
#         x = 1 + 1 # No yield = simulation clock stuck at current env.now

# ❌ BAD: Manual resource release (Dangerous)
def risky_request(env, res):
    req = res.request()
    yield req
    # If something fails here, the resource is never released
    res.release(req)

# ✅ GOOD: Using context manager
def safe_request(env, res):
    with res.request() as req:
        yield req
        # Resource automatically released when exiting 'with'

リソースと待ち行列（simpy.Resource）

共有容量の管理

def car(env, name, bcs, driving_time, charge_time):
    # Simulate driving to the station
    yield env.timeout(driving_time)

    print(f'{name} arriving at {env.now}')
    # Request one of the charging spots
    with bcs.request() as req:
        yield req
        print(f'{name} starting to charge at {env.now}')
        yield env.timeout(charge_time)
        print(f'{name} leaving at {env.now}')

env = simpy.Environment()
# Battery Charging Station with 2 spots
bcs = simpy.Resource(env, capacity=2)

for i in range(4):
    env.process(car(env, f'Car {i}', bcs, i*2, 5))

env.run()

高度なリソース：Container と Store

Container（液体／バルク）

# Modeling a gas station tank
gas_tank = simpy.Container(env, capacity=1000, init=500)

def refuel(env, gas_tank):
    yield gas_tank.put(400) # Add fuel

def car(env, gas_tank):
    yield gas_tank.get(40) # Consume fuel

Store（個別のオブジェクト）

# Modeling a producer-consumer buffer
buffer = simpy.Store(env, capacity=10)

def producer(env, buffer):
    yield buffer.put(f'Widget at {env.now}')

def consumer(env, buffer):
    item = yield buffer.get()
    print(f'Consumed {item}')

イベントの同期

複数イベントを待つ

# Wait for BOTH events
yield env.timeout(10) & other_process_event

# Wait for EITHER event
yield env.timeout(10) | other_process_event

実践的なワークフロー

1. 病院の救急外来モデル

def patient(env, name, nurse, doctor):
    arrival_time = env.now
    
    # 1. Wait for Nurse (Triage)
    with nurse.request() as req:
        yield req
        yield env.timeout(random.expovariate(1/5)) # 5 mins avg
        
    # 2. Wait for Doctor (Treatment)
    with doctor.request() as req:
        yield req
        yield env.timeout(random.normalvariate(20, 5)) # 20 mins avg
        
    wait_times.append(env.now - arrival_time)

wait_times = []
env = simpy.Environment()
nurse = simpy.Resource(env, capacity=2)
doctor = simpy.Resource(env, capacity=1)
# ... launch processes ...

2. 機械の故障と修理

def machine(env, repairman):
    while True:
        try:
            # Working phase
            working_time = 100
            yield env.timeout(working_time)
        except simpy.Interrupt:
            # Interrupted by failure
            with repairman.request() as req:
                yield req
                yield env.timeout(10) # Repair time

def break_machine(env, machine_proc):
    while True:
        yield env.timeout(random.uniform(50, 200))
        machine_proc.interrupt()

3. サーバーのリクエスト処理（データ収集）

import pandas as pd

log = []

def request(env, server):
    start_wait = env.now
    with server.request() as req:
        yield req
        wait = env.now - start_wait
        service_time = random.expovariate(1.0)
        yield env.timeout(service_time)
        
        log.append({
            'arrival': start_wait,
            'wait': wait,
            'service': service_time,
            'finish': env.now
        })

# After simulation:
# df = pd.DataFrame(log)
# print(df['wait'].mean())

パフォーマンスの最適化

simpy.PriorityResource の使用

重要度の異なるイベント（重症患者など）があるシステムでは、PriorityResource を使用してください。

res = simpy.PriorityResource(env, capacity=1)
with res.request(priority=0) as req: # Lower number = higher priority
    yield req

大規模シミュレーションの監視

高速シミュレーションでは、コンソールへの出力を避けてください。代わりにデータ収集用のリストを使用し、env.run() の完了後に NumPy／Pandas で処理してください。

よくある落とし穴と解決策

「停止したプロセス」

while True ループ内で yield が発生しないロジック分岐にコードが到達すると、SimPy は停止します。

# ❌ Problem:
def logic_error(env):
    while True:
        if condition:
            yield env.timeout(1)
        # Else? If condition is false, we loop forever at same time point!

# ✅ Solution:
def logic_fixed(env):
    while True:
        if condition:
            yield env.timeout(1)
        else:
            yield env.timeout(0.1) # Or some default wait

複数の環境

異なる環境のオブジェクト（リソース、プロセス）を混在させないでください。すべてのコンポーネントは、同じ simpy.Environment() インスタンスに属している必要があります。

割り込みのタイミング

プロセスに割り込みを行うと、現在の仮想時刻に割り込みが発生します。割り込まれたプロセスが何を yield していたかに注意してください（simpy.Interrupt 例外を受け取ります）。

ベストプラクティス

1. リソースには必ずコンテキストマネージャーを使用する - with resource.request() as req: を使うと、適切にクリーンアップできます
2. 再現性のために乱数シードを設定する - シミュレーションの開始時に random.seed(42) を使用します
3. データは外部に収集する - イベントデータは単なる print 文ではなく、リストや辞書に保存します
4. 明確な停止条件を定義する - 必ず具体的な時間制限を指定して env.run(until=...) を使用します
5. 仮想時間のみを使用する - time.sleep() や実時間を SimPy と混在させないでください
6. まず小規模なシミュレーションでテストする - 大規模なシミュレーションを実行する前にロジックを検証します
7. プロセスを文書化する - デバッグしやすいよう、プロセスとリソースには分かりやすい名前を付けます
8. 割り込みを適切に処理する - プロセスが割り込まれる可能性がある場合は、try/except ブロックを使用します

SimPy は Python のジェネレーター構文を、時間と希少性をモデル化する強力なエンジンに変えます。科学者やエンジニアは、実物を使った試験のコストやリスクを負うことなく、複雑なシステムについて「もしこうだったら？」というシナリオを試すことができます。