alterlab-simpy

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

概要

SimPy は、標準的な Python に基づくプロセスベースの離散イベントシミュレーションフレームワークです。エンティティ（顧客、車両、パケットなど）が互いに作用し、共有リソース（サーバー、機械、帯域幅など）を時間とともに奪い合うシステムのモデル化に SimPy を使用します。

主な機能：

• Python のジェネレーター関数を使用したプロセスのモデル化
• 共有リソースの管理（サーバー、コンテナー、ストア）
• イベント駆動型のスケジューリングと同期
• 実時間と同期するリアルタイムシミュレーション
• 包括的なモニタリングとデータ収集

このスキルを使用する場面

次の場合に SimPy スキルを使用します。

1. 離散イベントシステムのモデル化 - イベントが不規則な間隔で発生するシステム
2. リソースの競合 - エンティティが限られたリソース（サーバー、機械、スタッフ）を奪い合う場合
3. 待ち行列の分析 - 待ち行列、サービス時間、スループットの調査
4. プロセスの最適化 - 製造、物流、サービスプロセスの分析
5. ネットワークシミュレーション - パケットルーティング、帯域幅の割り当て、レイテンシの分析
6. キャパシティプランニング - 目標のパフォーマンスを実現する最適なリソースレベルの決定
7. システムの検証 - 実装前のシステム動作のテスト

適していない用途：

• 固定時間ステップでの連続シミュレーション（SciPy の ODE ソルバーを検討してください）
• リソースを共有しない独立したプロセス
• 純粋な数学的最適化（SciPy optimize を検討してください）

トリガーされないケース

シナリオ	代わりに使用
グリッドまたはネットワーク上の異質なエージェントを扱う社会システムのエージェントベースモデル（分離、意見の動態、拡散）	alterlab-abm-mesa
進化的アルゴリズムを使用したパレート最適な設計やパラメーターの探索（例：コストと待ち時間のトレードオフ）	alterlab-pymoo
観測された待ち行列やスループットデータに対する時系列モデルや回帰モデルの適合	alterlab-statsmodels
システムを制御する強化学習エージェントのトレーニング	alterlab-stable-baselines3

クイックスタート

基本的なシミュレーション構造

import simpy

def process(env, name):
    """A simple process that waits and prints."""
    print(f'{name} starting at {env.now}')
    yield env.timeout(5)
    print(f'{name} finishing at {env.now}')

# Create environment
env = simpy.Environment()

# Start processes
env.process(process(env, 'Process 1'))
env.process(process(env, 'Process 2'))

# Run simulation
env.run(until=10)

リソースの使用パターン

import simpy

def customer(env, name, resource):
    """Customer requests resource, uses it, then releases."""
    with resource.request() as req:
        yield req  # Wait for resource
        print(f'{name} got resource at {env.now}')
        yield env.timeout(3)  # Use resource
        print(f'{name} released resource at {env.now}')

env = simpy.Environment()
server = simpy.Resource(env, capacity=1)

env.process(customer(env, 'Customer 1', server))
env.process(customer(env, 'Customer 2', server))
env.run()

主な概念

1. 環境

シミュレーション環境が時間を管理し、イベントをスケジュールします。

import simpy

# Standard environment (runs as fast as possible)
env = simpy.Environment(initial_time=0)

# Real-time environment (synchronized with wall-clock)
import simpy.rt
env_rt = simpy.rt.RealtimeEnvironment(factor=1.0)

# Run simulation
env.run(until=100)  # Run until time 100
env.run()  # Run until no events remain

2. プロセス

プロセスは Python のジェネレーター関数（yield 文を含む関数）を使用して定義します。

def my_process(env, param1, param2):
    """Process that yields events to pause execution."""
    print(f'Starting at {env.now}')

    # Wait for time to pass
    yield env.timeout(5)

    print(f'Resumed at {env.now}')

    # Wait for another event
    yield env.timeout(3)

    print(f'Done at {env.now}')
    return 'result'

# Start the process
env.process(my_process(env, 'value1', 'value2'))

3. イベント

イベントはプロセスを同期する基本的な仕組みです。プロセスはイベントを yield し、そのイベントがトリガーされると再開します。

一般的なイベントの種類：

• env.timeout(delay) - 時間が経過するまで待機
• resource.request() - リソースを要求
• env.event() - カスタムイベントを作成
• env.process(func()) - プロセスをイベントとして扱う
• event1 & event2 - すべてのイベントを待機（AllOf）
• event1 | event2 - いずれかのイベントを待機（AnyOf）

リソース

SimPy は、さまざまなシナリオに対応する複数のリソースタイプを提供します。詳しくは references/resources.md を参照してください。

リソースタイプの概要

リソースタイプ	用途
Resource	容量に上限があるリソース（サーバー、機械）
PriorityResource	優先度に基づくキューイング
PreemptiveResource	高優先度のリクエストが低優先度の処理を中断可能
Container	バルク材料（燃料、水）
Store	Python オブジェクトの保管（FIFO）
FilterStore	アイテムの選択的な取得
PriorityStore	優先度順のアイテム

クイックリファレンス

import simpy

env = simpy.Environment()

# Basic resource (e.g., servers)
resource = simpy.Resource(env, capacity=2)

# Priority resource
priority_resource = simpy.PriorityResource(env, capacity=1)

# Container (e.g., fuel tank)
fuel_tank = simpy.Container(env, capacity=100, init=50)

# Store (e.g., warehouse)
warehouse = simpy.Store(env, capacity=10)

一般的なシミュレーションパターン

パターン 1：顧客とサーバーの待ち行列

import simpy
import random

def customer(env, name, server):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        print(f'{name} waited {wait:.2f}, served at {env.now}')
        yield env.timeout(random.uniform(2, 4))

def customer_generator(env, server):
    i = 0
    while True:
        yield env.timeout(random.uniform(1, 3))
        i += 1
        env.process(customer(env, f'Customer {i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(customer_generator(env, server))
env.run(until=20)

パターン 2：プロデューサーとコンシューマー

import simpy

def producer(env, store):
    item_id = 0
    while True:
        yield env.timeout(2)
        item = f'Item {item_id}'
        yield store.put(item)
        print(f'Produced {item} at {env.now}')
        item_id += 1

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

env = simpy.Environment()
store = simpy.Store(env, capacity=10)
env.process(producer(env, store))
env.process(consumer(env, store))
env.run(until=20)

パターン 3：並列タスクの実行

import simpy

def task(env, name, duration):
    print(f'{name} starting at {env.now}')
    yield env.timeout(duration)
    print(f'{name} done at {env.now}')
    return f'{name} result'

def coordinator(env):
    # Start tasks in parallel
    task1 = env.process(task(env, 'Task 1', 5))
    task2 = env.process(task(env, 'Task 2', 3))
    task3 = env.process(task(env, 'Task 3', 4))

    # Wait for all to complete
    results = yield task1 & task2 & task3
    print(f'All done at {env.now}')

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

ワークフローガイド

ステップ 1：システムを定義する

次の項目を特定します。

• エンティティ：システム内を移動するものは何ですか？（顧客、部品、パケット）
• リソース：制約となるものは何ですか？（サーバー、機械、帯域幅）
• プロセス：どのような活動がありますか？（到着、サービス、出発）
• 指標：何を測定しますか？（待ち時間、稼働率、スループット）

ステップ 2：プロセス関数を実装する

プロセスの種類ごとにジェネレーター関数を作成します。

def entity_process(env, name, resources, parameters):
    # Arrival logic
    arrival_time = env.now

    # Request resources
    with resource.request() as req:
        yield req

        # Service logic
        service_time = calculate_service_time(parameters)
        yield env.timeout(service_time)

    # Departure logic
    collect_statistics(env.now - arrival_time)

ステップ 3：モニタリングを設定する

モニタリングユーティリティを使用してデータを収集します。包括的な手法については、references/monitoring.md を参照してください。

from scripts.resource_monitor import ResourceMonitor

# Create and monitor resource
resource = simpy.Resource(env, capacity=2)
monitor = ResourceMonitor(env, resource, "Server")

# After simulation
monitor.report()

ステップ 4：実行して分析する

# Run simulation
env.run(until=simulation_time)

# Generate reports
monitor.report()
stats.report()

# Export data for further analysis
monitor.export_csv('results.csv')

高度な機能

プロセスの相互作用

プロセスは、イベント、プロセスの yield、割り込みを通じて相互に作用できます。詳しいパターンについては、references/process-interaction.md を参照してください。

主な仕組み：

• イベントシグナリング：調整に使用する共有イベント
• プロセスの yield：ほかのプロセスが完了するまで待機
• 割り込み：プリエンプションのためにプロセスを強制的に再開

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

ハードウェア・イン・ザ・ループやインタラクティブアプリケーション向けに、シミュレーションを実時間と同期させます。references/real-time.md を参照してください。

import simpy.rt

env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1:1 time mapping
# factor=0.5 means 1 sim unit = 0.5 seconds (2x faster)

包括的なモニタリング

プロセス、リソース、イベントを監視します。次の手法については、references/monitoring.md を参照してください。

• 状態変数の追跡
• リソースのモンキーパッチ
• イベントのトレース
• 統計情報の収集

スクリプトとテンプレート

basic_simulation_template.py

次の機能を備えた、キューシミュレーション構築用の完全なテンプレートです。

• 設定可能なパラメーター
• 統計情報の収集
• 顧客の生成
• リソースの使用状況
• レポートの生成

使用方法：

from scripts.basic_simulation_template import SimulationConfig, run_simulation

config = SimulationConfig()
config.num_resources = 2
config.sim_time = 100
stats = run_simulation(config)
stats.report()

resource_monitor.py

再利用可能なモニタリングユーティリティ：

• ResourceMonitor - 単一リソースを追跡
• MultiResourceMonitor - 複数のリソースを監視
• ContainerMonitor - コンテナーのレベルを追跡
• 統計情報を自動計算
• CSV エクスポート機能

使用方法：

from scripts.resource_monitor import ResourceMonitor

monitor = ResourceMonitor(env, resource, "My Resource")
# ... run simulation ...
monitor.report()
monitor.export_csv('data.csv')

インポートに関する注意： from scripts.... のパスは、スキルディレクトリからの相対パスです。スクリプトをシミュレーションの隣にコピーするか、インポートする前にスキルディレクトリを PYTHONPATH に追加してください。

リファレンスドキュメント

必要に応じて読み込める詳細ガイド：

• references/resources.md - すべてのリソースの種類と例
• references/events.md - イベントシステムとパターン
• references/process-interaction.md - プロセスの同期（シグナル、yield、割り込み）
• references/monitoring.md - データ収集の手法
• references/real-time.md - リアルタイムシミュレーションの設定

ベストプラクティス

1. ジェネレーター関数：プロセス関数では必ず yield を使用します
2. リソースのコンテキストマネージャー：自動クリーンアップのため、with resource.request() as req: を使用します
3. 再現性：一貫した結果を得るために random.seed() を設定します
4. モニタリング：シミュレーション終了時だけでなく、実行中もデータを収集します
5. 検証：単純なケースを解析解と比較します
6. ドキュメント：プロセスのロジックとパラメーターの選択理由をコメントします
7. モジュール設計：プロセスのロジック、統計情報、設定を分離します

よくある落とし穴

1. yield の付け忘れ：プロセスを一時停止するには、イベントを yield する必要があります
2. イベントの再利用：イベントをトリガーできるのは一度だけです
3. リソースリーク：コンテキストマネージャーを使用するか、確実に解放します
4. ブロッキング操作：プロセス内で Python のブロッキング呼び出しを避けます
5. 時間単位：時間単位の解釈に一貫性を持たせます
6. デッドロック：少なくとも1つのプロセスが進行できることを確認します

ユースケースの例

• 製造：機械のスケジューリング、生産ライン、在庫管理
• 医療：救急外来のシミュレーション、患者の流れ、スタッフの割り当て
• 通信：ネットワークトラフィック、パケットルーティング、帯域幅の割り当て
• 輸送：交通の流れ、物流、車両のルーティング
• サービス業務：コールセンター、小売店のレジ、予約のスケジューリング
• コンピューターシステム：CPU スケジューリング、メモリ管理、I/O 操作

AlterLab Academic Skills スイートの一部です。