基于Python Simpy的蒙特卡洛仿真：生产线Weibull参数动态调整问题

你的核心问题在于状态跟踪逻辑混乱和失效参数动态调整的触发方式错误，原代码用Container轮询的方式不仅效率低，还容易导致状态判断错误。下面是修正后的实现，完全贴合你的需求：

核心改进点 #

• 直接跟踪两台生产线的运行状态，替代原有的Container方案
• 机器失效时间根据当前是否处于提速状态，动态选择Weibull分布的scale参数
• 用事件触发替代轮询，状态变更时即时更新参数
• 简化状态变量，避免冗余判断

修正后的代码 #

import simpy
import numpy as np
from scipy.stats import weibull_min

class ProductionLine:
    def __init__(self, env, name, normal_scale, speedup_scale, shape, mean_repair):
        self.env = env
        self.name = name
        self.normal_scale = normal_scale  # 正常状态下的Weibull scale
        self.speedup_scale = speedup_scale  # 提速状态下的Weibull scale
        self.shape = shape
        self.mean_repair = mean_repair
        self.is_running = True  # 是否正常运行
        self.is_speedup = False  # 是否处于提速状态
        
        # 共享状态：记录当前正常运行的生产线数量
        self.shared_running_count = env.shared_running_count
        
        # 启动生产线主流程
        self.process = env.process(self.run())
        
    def generate_ttf(self):
        # 根据当前状态选择对应的scale参数生成失效时间
        scale = self.speedup_scale if self.is_speedup else self.normal_scale
        ttf = weibull_min.rvs(self.shape, loc=0, scale=scale)
        # 确保失效时间不为0
        return max(ttf, 0.1)
    
    def run(self):
        while True:
            # 生成当前状态下的失效时间
            ttf = self.generate_ttf()
            try:
                # 等待失效发生
                yield self.env.timeout(ttf)
                # 生产线故障
                self.is_running = False
                self.shared_running_count -= 1
                print(f"[{self.env.now:.2f}] {self.name} 故障，剩余正常生产线: {self.shared_running_count.value}")
                
                # 通知其他生产线切换状态（如果只剩一条）
                if self.shared_running_count.value == 1:
                    for line in self.env.lines:
                        if line.is_running:
                            line.is_speedup = True
                            print(f"[{self.env.now:.2f}] {line.name} 切换至提速模式")
                
                # 修复时间
                repair_time = np.random.exponential(self.mean_repair)
                yield self.env.timeout(repair_time)
                
                # 生产线修复完成
                self.is_running = True
                self.shared_running_count += 1
                print(f"[{self.env.now:.2f}] {self.name} 修复完成，当前正常生产线: {self.shared_running_count.value}")
                
                # 通知其他生产线恢复正常模式（如果两条都正常）
                if self.shared_running_count.value == 2:
                    for line in self.env.lines:
                        line.is_speedup = False
                        print(f"[{self.env.now:.2f}] {line.name} 切换至正常模式")
                        
            except simpy.Interrupt:
                # 状态变更通过is_speedup变量即时生效，无需处理中断
                pass

# 初始化仿真环境
env = simpy.Environment()
# 共享变量：记录当前正常运行的生产线数量
env.shared_running_count = simpy.Store(env, capacity=1)
env.shared_running_count.put(2)
# 存储所有生产线实例，方便状态通知
env.lines = []

# 创建两条生产线：正常scale=12，提速scale=10，shape=0.7，平均修复时间5分钟
line_a = ProductionLine(env, "Line A", normal_scale=12, speedup_scale=10, shape=0.7, mean_repair=5)
line_b = ProductionLine(env, "Line B", normal_scale=12, speedup_scale=10, shape=0.7, mean_repair=5)
env.lines.extend([line_a, line_b])

# 运行仿真60分钟
env.run(until=60)

代码解释 #

1. 状态跟踪：用simpy.Store维护当前正常运行的生产线数量，生产线故障/修复时即时更新这个值
2. 动态参数调整：generate_ttf方法根据is_speedup状态选择对应的scale参数，每次生成失效时间前都会检查当前状态，确保参数正确
3. 状态触发：当生产线数量变为1时，自动将剩余生产线切换为提速模式；恢复为2时，切换回正常模式，无需低效轮询
4. 失效与修复逻辑：严格按照Weibull分布生成失效时间，修复时间可根据需求替换为你需要的对数正态分布

原代码问题总结 #

• 用容量3的Container跟踪状态完全不符合两条生产线的需求，逻辑冗余
• check_machines_status的轮询逻辑会频繁触发不必要的中断，导致流程混乱
• updated_mean_time_to_failure错误修改了shape参数，而需求是调整scale参数
• 冗余的状态变量（broken和machine_status）导致逻辑冲突

内容的提问来源于stack exchange，提问作者Jai Pandey