离散事件仿真器Skill discrete-event-simulator

---

name: discrete-event-simulator description: 用于建模和分析具有随机过程的复杂系统的离散事件仿真技能。 allowed-tools: Bash(*) 读取 写入 编辑 全局搜索 Grep 网络获取 metadata: author: babysitter-sdk version: “1.0.0” category: 仿真 backlog-id: SK-IE-005

离散事件仿真器

您是 离散事件仿真器 - 一个专门用于为具有随机过程的复杂系统构建和分析离散事件仿真模型的技能。

概述

此技能支持AI驱动的离散事件仿真，包括：

• 使用SimPy进行流程建模
• 使用统计分布生成实体
• 资源容量建模
• 队列规则实现（FIFO、优先级等）
• 仿真预热期检测
• 带置信区间的输出统计
• 动画和可视化生成

先决条件

• Python 3.8+ 并安装SimPy
• 统计库（scipy, numpy）
• 可视化库（matplotlib, plotly）

能力

1. 基础SimPy模型

import simpy
import numpy as np

def manufacturing_system(env, arrival_rate, service_rate, num_machines):
    """
    简单制造系统仿真
    """
    machines = simpy.Resource(env, capacity=num_machines)

    # 统计收集
    wait_times = []
    system_times = []

    def customer(env, name, machines):
        arrival_time = env.now

        with machines.request() as request:
            yield request
            wait_time = env.now - arrival_time
            wait_times.append(wait_time)

            # 服务时间
            service_time = np.random.exponential(1/service_rate)
            yield env.timeout(service_time)

            system_times.append(env.now - arrival_time)

    def customer_generator(env, arrival_rate, machines):
        customer_id = 0
        while True:
            yield env.timeout(np.random.exponential(1/arrival_rate))
            customer_id += 1
            env.process(customer(env, f"Customer_{customer_id}", machines))

    env.process(customer_generator(env, arrival_rate, machines))
    env.run(until=1000)

    return {
        "avg_wait_time": np.mean(wait_times),
        "avg_system_time": np.mean(system_times),
        "max_wait_time": np.max(wait_times),
        "customers_served": len(wait_times)
    }

2. 复杂流程

class ManufacturingLine:
    """
    带缓冲区的多阶段制造流水线
    """
    def __init__(self, env, config):
        self.env = env
        self.config = config

        # 创建资源
        self.stations = {
            name: simpy.Resource(env, capacity=cap)
            for name, cap in config['stations'].items()
        }

        # 工位间缓冲区
        self.buffers = {
            name: simpy.Container(env, capacity=cap, init=0)
            for name, cap in config['buffers'].items()
        }

        # 统计
        self.stats = {
            'throughput': 0,
            'wip': [],
            'utilization': {s: [] for s in self.stations}
        }

    def part_flow(self, part_id):
        """零件流经所有工位的处理过程"""
        for station_name in self.config['routing']:
            station = self.stations[station_name]

            with station.request() as req:
                yield req

                # 记录利用率
                self.stats['utilization'][station_name].append(
                    station.count / station.capacity
                )

                # 处理时间
                process_time = self.config['process_times'][station_name]()
                yield self.env.timeout(process_time)

        self.stats['throughput'] += 1

    def part_arrivals(self):
        """生成到达的零件"""
        part_id = 0
        while True:
            yield self.env.timeout(self.config['interarrival_time']())
            part_id += 1
            self.env.process(self.part_flow(part_id))
            self.stats['wip'].append(sum(s.count for s in self.stations.values()))

3. 队列规则

class PriorityQueue:
    """
    SimPy的优先级队列资源
    """
    def __init__(self, env, capacity):
        self.env = env
        self.capacity = capacity
        self.queue = []
        self.available = capacity

    def request(self, priority=0):
        return PriorityRequest(self, priority)

class PriorityRequest:
    def __init__(self, resource, priority):
        self.resource = resource
        self.priority = priority
        self.event = resource.env.event()

    def __enter__(self):
        if self.resource.available > 0:
            self.resource.available -= 1
            self.event.succeed()
        else:
            # 按优先级插入（数值越小优先级越高）
            heapq.heappush(self.resource.queue,
                          (self.priority, self.event))
        return self.event

    def __exit__(self, *args):
        self.resource.available += 1
        if self.resource.queue:
            _, waiting_event = heapq.heappop(self.resource.queue)
            waiting_event.succeed()

4. 预热期检测

def welch_warmup_detection(data, window_size=50):
    """
    用于检测预热期的Welch方法
    """
    n = len(data)
    moving_avg = []

    for i in range(n - window_size + 1):
        moving_avg.append(np.mean(data[i:i+window_size]))

    # 寻找收敛点
    threshold = 0.01 * np.std(moving_avg)
    converged = False
    warmup_end = 0

    for i in range(len(moving_avg) - 1):
        if abs(moving_avg[i+1] - moving_avg[i]) < threshold:
            if not converged:
                warmup_end = i
                converged = True
        else:
            converged = False

    return warmup_end * window_size

def run_with_warmup(env, model, warmup_time, run_time):
    """
    带预热期移除的仿真运行
    """
    env.run(until=warmup_time)
    model.reset_statistics()
    env.run(until=warmup_time + run_time)
    return model.get_statistics()

5. 带置信区间的输出分析

def replicated_runs(model_func, num_replications, confidence=0.95):
    """
    运行多次重复实验并计算置信区间
    """
    results = []

    for rep in range(num_replications):
        env = simpy.Environment()
        result = model_func(env, seed=rep)
        results.append(result)

    # 计算统计量
    means = {key: np.mean([r[key] for r in results])
             for key in results[0].keys()}

    stds = {key: np.std([r[key] for r in results], ddof=1)
            for key in results[0].keys()}

    # 置信区间
    from scipy import stats
    t_value = stats.t.ppf((1 + confidence) / 2, num_replications - 1)

    ci = {key: (means[key] - t_value * stds[key] / np.sqrt(num_replications),
                means[key] + t_value * stds[key] / np.sqrt(num_replications))
          for key in means.keys()}

    return {
        "means": means,
        "std_devs": stds,
        "confidence_intervals": ci,
        "num_replications": num_replications
    }

流程集成

此技能与以下流程集成：

• discrete-event-simulation-modeling.js
• queuing-system-analysis.js
• capacity-planning-analysis.js

输出格式

{
  "model_name": "制造流水线",
  "simulation_time": 10000,
  "replications": 30,
  "warmup_time": 1000,
  "performance_measures": {
    "throughput": {
      "mean": 245.3,
      "std": 12.4,
      "ci_95": [240.1, 250.5]
    },
    "avg_wait_time": {
      "mean": 5.2,
      "std": 0.8,
      "ci_95": [4.9, 5.5]
    },
    "utilization": {
      "station_1": 0.82,
      "station_2": 0.91
    }
  },
  "bottleneck": "station_2",
  "recommendations": [
    "考虑在station_2增加产能"
  ]
}

工具/库

库	描述	使用场景
SimPy	DES框架	通用仿真
Ciw	队列网络	服务系统
salabim	动画	可视化模型
scipy.stats	统计	输出分析

最佳实践

1. 用解析结果验证 - 与已知解对比测试
2. 消除预热偏差 - 使用适当的初始化
3. 运行足够的重复实验 - 目标为窄置信区间
4. 记录假设 - 记录所有分布
5. 验证随机流 - 使用独立种子
6. 仔细分析输出 - 检查非平稳性

约束

• 报告置信区间，而不仅仅是点估计
• 记录所有随机数流
• 逐步验证模型行为
• 使用适当的运行时长