simpy-discrete-event-simulation

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

概要

SimPy は、標準的な Python ジェネレーターを使用するプロセスベースの離散イベントシミュレーションフレームワークです。エンティティ（顧客、車両、パケット）が共有リソース（サーバー、機械、帯域幅）と時間の経過とともに相互作用するシステムをモデル化します。イベント駆動型のスケジューリングと、任意でリアルタイム同期に対応しています。

使用する場面

• リソースの競合が発生する待ち行列システム（サーバー、機械、スタッフ）のモデル化
• 製造プロセスのシミュレーション（生産ライン、スケジューリング、ボトルネック分析）
• ネットワークシミュレーション（パケットルーティング、帯域幅の割り当て、レイテンシー分析）
• キャパシティプランニング（目標スループットを達成する最適なリソースレベルの決定）
• 医療業務（救急患者の流れ、スタッフの配置、病床管理）
• 物流と輸送（倉庫業務、車両ルーティング）
• 連続時間の ODE システムには → SciPy solve_ivp を使用してください
• エージェントベースモデリングには → Mesa を使用してください

前提条件

# pip install simpy
import simpy
import random

クイックスタート

import simpy
import random

def customer(env, name, server):
    """Customer arrives, waits for server, gets served, departs."""
    arrival = env.now
    with server.request() as req:
        yield req  # Wait in queue
        wait = env.now - arrival
        yield env.timeout(random.expovariate(1/3))  # Service time
        print(f'{name}: waited {wait:.1f}, served at {env.now:.1f}')

def arrivals(env, server):
    for i in range(20):
        yield env.timeout(random.expovariate(1/2))  # Inter-arrival
        env.process(customer(env, f'C{i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(arrivals(env, server))
env.run(until=50)

コア API

1. 環境とプロセス

import simpy

# Standard environment
env = simpy.Environment(initial_time=0)

# Processes are Python generators that yield events
def machine(env, name, repair_time):
    while True:
        yield env.timeout(random.expovariate(1/10))  # Time to failure
        print(f'{name} broke at {env.now:.1f}')
        yield env.timeout(repair_time)
        print(f'{name} repaired at {env.now:.1f}')

# Start processes — returns a Process event
proc = env.process(machine(env, 'Machine-1', repair_time=2))

# Run until time limit or no events remain
env.run(until=100)
# env.run()  # Run until no more events

# Current simulation time
print(f'Final time: {env.now}')

# Processes can return values and be awaited
def subtask(env, duration):
    yield env.timeout(duration)
    return f'completed in {duration}'

def main_task(env):
    # Sequential: wait for one process
    result = yield env.process(subtask(env, 5))
    print(f'Subtask {result} at {env.now}')

    # Parallel: wait for ALL (AllOf)
    t1 = env.process(subtask(env, 3))
    t2 = env.process(subtask(env, 4))
    results = yield t1 & t2  # AllOf — resumes when both done
    print(f'Both done at {env.now}')

    # Race: wait for ANY (AnyOf)
    t3 = env.process(subtask(env, 2))
    t4 = env.process(subtask(env, 6))
    result = yield t3 | t4  # AnyOf — resumes when first completes
    print(f'First done at {env.now}')

env = simpy.Environment()
env.process(main_task(env))
env.run()

2. リソース

import simpy

env = simpy.Environment()

# Basic resource — capacity-limited (e.g., 2 servers)
server = simpy.Resource(env, capacity=2)
print(f'Capacity: {server.capacity}, In use: {server.count}, Queue: {len(server.queue)}')

# Priority resource — lower number = higher priority
priority_server = simpy.PriorityResource(env, capacity=1)

def vip_customer(env, res):
    with res.request(priority=1) as req:  # Higher priority
        yield req
        yield env.timeout(3)

def regular_customer(env, res):
    with res.request(priority=10) as req:  # Lower priority
        yield req
        yield env.timeout(3)

# Preemptive resource — high priority interrupts low priority
preemptive = simpy.PreemptiveResource(env, capacity=1)

def urgent_job(env, res):
    with res.request(priority=0, preempt=True) as req:
        yield req  # May interrupt current user
        yield env.timeout(1)

# Container — bulk material (fuel, water, inventory)
tank = simpy.Container(env, capacity=100, init=50)

def refuel(env, tank):
    yield tank.put(30)  # Add 30 units
    print(f'Tank level: {tank.level}/{tank.capacity}')

def consume(env, tank):
    yield tank.get(20)  # Remove 20 units
    print(f'Tank level: {tank.level}/{tank.capacity}')

# Store — FIFO object storage
warehouse = simpy.Store(env, capacity=10)

def producer(env, store):
    for i in range(5):
        yield env.timeout(2)
        yield store.put(f'Item-{i}')

def consumer(env, store):
    while True:
        item = yield store.get()
        print(f'Got {item} at {env.now}')
        yield env.timeout(3)

# FilterStore — selective retrieval
parts = simpy.FilterStore(env, capacity=20)

def picker(env, store):
    # Get specific item matching condition
    item = yield store.get(lambda x: x['color'] == 'red')
    print(f'Found red item: {item}')

3. イベントと同期

import simpy

env = simpy.Environment()

# Basic event — manual trigger for signaling between processes
signal = env.event()

def waiter(env, event):
    print(f'Waiting at {env.now}')
    value = yield event  # Blocks until triggered
    print(f'Got signal "{value}" at {env.now}')

def sender(env, event):
    yield env.timeout(5)
    event.succeed(value='go')  # Trigger with value

env.process(waiter(env, signal))
env.process(sender(env, signal))
env.run()
# Output: Waiting at 0, Got signal "go" at 5

# Timeout — most common event
yield env.timeout(delay=5)

# Process interruption
def interruptible(env, name):
    try:
        yield env.timeout(10)
    except simpy.Interrupt as interrupt:
        print(f'{name} interrupted: {interrupt.cause} at {env.now}')

def interruptor(env, proc):
    yield env.timeout(3)
    proc.interrupt('maintenance')

proc = env.process(interruptible(env, 'Worker'))
env.process(interruptor(env, proc))

# Barrier synchronization — wait for N processes
class Barrier:
    def __init__(self, env, n):
        self.env = env
        self.n = n
        self.count = 0
        self.event = env.event()

    def wait(self):
        self.count += 1
        if self.count >= self.n:
            self.event.succeed()
        return self.event

def phase_worker(env, name, barrier):
    yield env.timeout(random.uniform(1, 5))  # Phase work
    print(f'{name} reached barrier at {env.now:.1f}')
    yield barrier.wait()  # Wait for all workers
    print(f'{name} passed barrier at {env.now:.1f}')

env = simpy.Environment()
barrier = Barrier(env, n=3)
for i in range(3):
    env.process(phase_worker(env, f'W{i}', barrier))
env.run()

4. モニタリングと統計

import simpy

# Inline statistics collection
class Stats:
    def __init__(self):
        self.wait_times = []
        self.queue_lengths = []

    def report(self):
        if self.wait_times:
            avg_wait = sum(self.wait_times) / len(self.wait_times)
            max_wait = max(self.wait_times)
            print(f'Avg wait: {avg_wait:.2f}, Max wait: {max_wait:.2f}')
            print(f'Customers served: {len(self.wait_times)}')

def customer(env, name, server, stats):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        stats.wait_times.append(wait)
        stats.queue_lengths.append(len(server.queue))
        yield env.timeout(random.expovariate(1/3))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
stats = Stats()

def gen(env, server, stats):
    for i in range(100):
        yield env.timeout(random.expovariate(1/2))
        env.process(customer(env, f'C{i}', server, stats))

env.process(gen(env, server, stats))
env.run(until=200)
stats.report()

# Resource monitoring via monkey-patching
def patch_resource(resource, data):
    """Patch resource to log request/release events."""
    original_request = resource.request
    original_release = resource.release

    def monitored_request(*args, **kwargs):
        req = original_request(*args, **kwargs)
        data.append((resource._env.now, 'request', resource.count, len(resource.queue)))
        return req

    def monitored_release(*args, **kwargs):
        result = original_release(*args, **kwargs)
        data.append((resource._env.now, 'release', resource.count, len(resource.queue)))
        return result

    resource.request = monitored_request
    resource.release = monitored_release

log = []
patch_resource(server, log)
# After simulation: analyze log for utilization, queue dynamics

5. リアルタイムシミュレーション

import simpy.rt

# Real-time environment — synchronized with wall clock
env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1 sim unit = 1 second
# factor=0.1 → 10x faster (1 sim unit = 0.1 seconds)
# factor=60 → 1 sim unit = 1 minute

# Strict mode raises RuntimeError if simulation can't keep up
env_strict = simpy.rt.RealtimeEnvironment(factor=1.0, strict=True)

# Non-strict mode (default) allows slower-than-real-time execution
env_relaxed = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)

def periodic_task(env, interval):
    while True:
        print(f'Tick at sim time {env.now:.1f}')
        yield env.timeout(interval)

env = simpy.rt.RealtimeEnvironment(factor=1.0)
env.process(periodic_task(env, 2.0))
env.run(until=10)
# Prints "Tick" every ~2 real seconds

重要な概念

リソース選択ガイド

必要な用途	リソースタイプ	主な機能
サーバー／機械の数に制限がある	Resource	FIFO キュー、容量制限
優先度付きキューイング	PriorityResource	数値が小さいほど優先度が高い
プリエンプティブスケジューリング	PreemptiveResource	優先度の高い処理が現在の利用者を中断
バルク材料（燃料、水）	Container	put(amount) ／ get(amount)、連続的な残量
オブジェクトキュー（FIFO）	Store	put(item) ／ get()、順序に従った取得
条件付き取得	FilterStore	get(lambda x: condition)
優先度順のアイテム	PriorityStore	優先度に従ってアイテムを並べ替え

プロセス間の相互作用メカニズム

メカニズム	使用する場面	コードパターン
イベントシグナリング	複数の待機プロセスへのブロードキャスト	event = env.event() → yield event ／ event.succeed()
プロセスの yield	順次または並列実行	yield env.process(func()) または yield p1 & p2
割り込み	プリエンプション、メンテナンス、キャンセル	proc.interrupt(cause) + try/except simpy.Interrupt
タイムアウトとの競合	キャンセル可能なタイムアウト	`yield event

よくあるワークフロー

1. 製造ラインのシミュレーション

import simpy
import random

def part(env, name, machines, buffer, stats):
    """Part flows through sequential machines with intermediate buffer."""
    for i, machine in enumerate(machines):
        with machine.request() as req:
            yield req
            process_time = random.triangular(1, 3, 2)
            yield env.timeout(process_time)

    if buffer.level < buffer.capacity:
        yield buffer.put(1)
        stats['produced'] += 1

def part_generator(env, machines, buffer, stats):
    i = 0
    while True:
        yield env.timeout(random.expovariate(1/2))
        env.process(part(env, f'Part-{i}', machines, buffer, stats))
        i += 1

random.seed(42)
env = simpy.Environment()
machines = [simpy.Resource(env, capacity=1) for _ in range(3)]
output_buffer = simpy.Container(env, capacity=100, init=0)
stats = {'produced': 0}
env.process(part_generator(env, machines, output_buffer, stats))
env.run(until=480)  # 8-hour shift
print(f'Parts produced: {stats["produced"]}')
print(f'Buffer level: {output_buffer.level}')

2. 優先度付きのマルチサーバーキュー

import simpy
import random

def patient(env, name, priority, er, stats):
    arrival = env.now
    with er.request(priority=priority) as req:
        yield req
        wait = env.now - arrival
        stats['waits'].append((name, priority, wait))
        service = random.expovariate(1/15)  # ~15 min avg
        yield env.timeout(service)

def patient_arrivals(env, er, stats):
    i = 0
    while True:
        yield env.timeout(random.expovariate(1/5))  # ~5 min between arrivals
        pri = random.choices([1, 2, 3], weights=[0.1, 0.3, 0.6])[0]
        env.process(patient(env, f'P{i}', pri, er, stats))
        i += 1

random.seed(42)
env = simpy.Environment()
er = simpy.PriorityResource(env, capacity=3)
stats = {'waits': []}
env.process(patient_arrivals(env, er, stats))
env.run(until=480)

# Analyze by priority
for pri in [1, 2, 3]:
    waits = [w for _, p, w in stats['waits'] if p == pri]
    if waits:
        print(f'Priority {pri}: avg wait {sum(waits)/len(waits):.1f}, n={len(waits)}')

3. モニタリング付きのプロデューサー・コンシューマー

テキストのみのワークフロー（コア API のモジュール 2、3、4 を組み合わせます）：

1. 容量を制限した simpy.Store を作成します（モジュール 2：リソース）
2. 生産遅延を伴い、yield store.put(item) を実行するプロセスを実装します（モジュール 2）
3. 処理遅延を伴い、yield store.get() を実行するコンシューマープロセスを実装します（モジュール 2）
4. ストアが満杯になったときのバックプレッシャーに対応するイベントシグナリングを追加します（モジュール 3：イベント）
5. スループット、キューの長さ、アイドル時間の統計を収集します（モジュール 4：モニタリング）
6. シミュレーションを実行し、レポートを生成します

主なパラメーター

パラメーター	モジュール	デフォルト	範囲	効果
capacity	Resource	1	1–∞	同時に利用できるユーザー数
priority	PriorityResource.request	0	int	数値が小さいほど優先度が高い
preempt	PreemptiveResource.request	True	bool	優先度の低い処理を中断するかどうか
capacity	Container	float('inf')	0–∞	最大容量
init	Container	0	0–capacity	初期残量
capacity	Store	float('inf')	0–∞	最大アイテム数
factor	RealtimeEnvironment	1.0	>0	シミュレーション時間と実時間の比率
strict	RealtimeEnvironment	False	bool	スケジュールに遅れた場合にエラーを発生させるかどうか
initial_time	Environment	0	any float	シミュレーション開始時刻

ベストプラクティス

1. リソースには必ずコンテキストマネージャーを使用します：with resource.request() as req: yield req によって、例外が発生した場合もリソースが自動的に解放されます
2. 再現性を確保するために乱数シードを設定します：プロセスを作成する前に random.seed(42) を実行します。より多くの分布を扱う場合は numpy.random を使用します
3. シミュレーション中に統計を収集します：シミュレーション中にリストへ追加し、集計は env.run() の後に計算します。シミュレーション中に問い合わせないでください
4. プロセス時間には三角分布を使用します：random.triangular(min, max, mode) はサービス時間を表す場合、一様分布よりも現実的です
5. アンチパターン — yield の付け忘れ：yield を付けずに env.timeout(5) を実行すると、イベントは作成されますが、プロセスは一時停止しません。必ず yield env.timeout(5) としてください
6. アンチパターン — イベントの再利用：イベントは一度しかトリガーできません。シグナルサイクルごとに新しい env.event() を作成します。繰り返し可能なシグナルでは、ループ内で新しいイベントを作成します

よく使うレシピ

レシピ：複数回の反復を伴うシミュレーション

import simpy
import random
import statistics

def run_single(seed, sim_time=480, n_servers=2):
    random.seed(seed)
    env = simpy.Environment()
    server = simpy.Resource(env, capacity=n_servers)
    waits = []

    def customer(env, server):
        arrival = env.now
        with server.request() as req:
            yield req
            waits.append(env.now - arrival)
            yield env.timeout(random.expovariate(1/3))

    def gen(env, server):
        while True:
            yield env.timeout(random.expovariate(1/2))
            env.process(customer(env, server))

    env.process(gen(env, server))
    env.run(until=sim_time)
    return sum(waits) / len(waits) if waits else 0

# Run 30 replications
results = [run_single(seed=i) for i in range(30)]
print(f'Mean avg wait: {statistics.mean(results):.2f}')
print(f'95% CI: ±{1.96 * statistics.stdev(results) / len(results)**0.5:.2f}')

レシピ：割り込みベースのメンテナンス

import simpy
import random

def machine(env, name, repair_crew):
    while True:
        try:
            # Operate until failure
            ttf = random.expovariate(1/50)  # Mean 50 time units to failure
            yield env.timeout(ttf)
            print(f'{name} failed at {env.now:.1f}')
        except simpy.Interrupt:
            print(f'{name} interrupted for maintenance at {env.now:.1f}')

        # Repair (needs repair crew)
        with repair_crew.request() as req:
            yield req
            repair = random.uniform(2, 5)
            yield env.timeout(repair)
            print(f'{name} repaired at {env.now:.1f}')

def maintenance_scheduler(env, machines_procs):
    """Periodic preventive maintenance every 40 time units."""
    while True:
        yield env.timeout(40)
        for proc in machines_procs:
            if proc.is_alive:
                proc.interrupt('scheduled maintenance')

env = simpy.Environment()
repair_crew = simpy.Resource(env, capacity=1)
procs = [env.process(machine(env, f'M{i}', repair_crew)) for i in range(3)]
env.process(maintenance_scheduler(env, procs))
env.run(until=200)

レシピ：コンテナベースのサプライチェーン

import simpy
import random

def supplier(env, warehouse):
    """Deliver batch when level drops below reorder point."""
    while True:
        if warehouse.level < 20:  # Reorder point
            yield env.timeout(random.uniform(5, 10))  # Lead time
            amount = min(50, warehouse.capacity - warehouse.level)
            yield warehouse.put(amount)
            print(f'Delivered {amount} units at {env.now:.1f}, level={warehouse.level}')
        yield env.timeout(1)  # Check interval

def demand(env, warehouse, stats):
    while True:
        yield env.timeout(random.expovariate(1/2))
        qty = random.randint(1, 5)
        if warehouse.level >= qty:
            yield warehouse.get(qty)
            stats['fulfilled'] += qty
        else:
            stats['stockouts'] += 1

env = simpy.Environment()
warehouse = simpy.Container(env, capacity=100, init=80)
stats = {'fulfilled': 0, 'stockouts': 0}
env.process(supplier(env, warehouse))
env.process(demand(env, warehouse, stats))
env.run(until=500)
print(f'Fulfilled: {stats["fulfilled"]}, Stockouts: {stats["stockouts"]}')

トラブルシューティング

問題	原因	解決策
プロセスが一時停止しない	イベントの前に yield がない	env.timeout(x) だけでなく、必ず yield env.timeout(x) とします
RuntimeError: Event already triggered	トリガー済みイベントの再利用	シグナルサイクルごとに新しい env.event() を作成します
リソースが解放されない	コンテキストマネージャーを使用していない	with resource.request() as req: のパターンを使用します
シミュレーションが永久に実行される	until パラメーターがなく、プロセスが無限に続く	env.run(until=time) を追加するか、プロセスが終了するようにします
simpy.Interrupt が捕捉されない	割り込み可能なプロセスに try/except がない	yield を try: ... except simpy.Interrupt: で囲みます
キューの順序が誤っている	PriorityResource ではなく Resource を使用している	優先度付きキューには simpy.PriorityResource に切り替えます
リアルタイム実行が遅すぎる	計算が実時間の予算を超えている	strict=False を設定するか、factor を大きくします
コンテナの put がブロックされる	コンテナが容量上限に達している	put の前に container.level < container.capacity を確認します
FilterStore の get が永久にブロックされる	条件に一致するアイテムがない	生産側でフィルター条件に一致するアイテムが作成されることを確認します
統計が空になっている	env.run() の前に収集している	env.run() の完了後に stats.report() を呼び出します

同梱リソース

• references/process_events_guide.md — イベントのライフサイクル（トリガー済み→処理済み）、複合イベント（AllOf/AnyOf）、プロセス間の相互作用パターン（シグナリング、バリア、割り込み、ハンドシェイク）、高度な同期について詳しく説明します。元の events.md（375 行）と process-interaction.md（425 行）を統合したものです
• references/resources_monitoring_guide.md — リソースの種類に関する完全なリファレンス（Resource、Priority、Preemptive、Container、Store、FilterStore、PriorityStore）、モンキーパッチによるモニタリング（ResourceMonitor、ContainerMonitor クラス）、統計の収集パターン、CSV/matplotlib へのエクスポート、リアルタイムシミュレーション（RealtimeEnvironment、時間スケーリング、strict モード、HIL パターン）について説明します。元の resources.md（276 行）、monitoring.md（476 行）、real-time.md（396 行）を統合したものです。スクリプトの機能（basic_simulation_template.py、resource_monitor.py）は、Core API のモニタリング例とよく使うレシピに取り込まれています

関連スキル

• matplotlib-scientific-plotting — シミュレーション結果（キューの長さ、時間経過に伴う利用率）を可視化します
• polars-dataframes — 大規模なシミュレーション出力データセットを分析します

参考資料

• SimPy のドキュメント：https://simpy.readthedocs.io/
• SimPy の GitHub：https://github.com/teamhide/simpy
• Banks ほか、『離散事象システムシミュレーション』（教科書の参考文献）