python的智能制造导论工业场景模拟第一百一十五篇:搭建离散事件仿真，模拟多品种混线生产，统计在制品库存，评估不同排产策略库存压力。

离散事件仿真多品种混线生产统计在制品库存评估不同排产策略库存压力周四上午生产计划员小刘把三张工单甩在会议桌上脸色不太好看。这个月第三条线要同时跑A、B、C三种产品订单量分别是A 500件、B 300件、C 200件。小刘指着排产表问题是——A的工装夹具和B、C完全不通用换型一次要45分钟。如果按先A后B再C的顺序排换型两次总共90分钟但如果穿插着排换型次数直接翻倍光换型就浪费半天。那就别穿插啊按产品族集中排呗车间主任说。集中排小刘摇头A产品客户要求月底前交付B和C是下个月。如果先集中跑完AB和C的物料就得在产线上堆着等——在制品库存WIP直接爆仓。仓库那边昨天刚找我吵了一架说缓存区已经堆了三天了。那你们MES系统不能模拟一下吗我问。MES是执行系统不是仿真系统。小刘苦笑它只能记录实际发生了什么不能回答如果按方案X排产WIP会是多少。我需要的是在排产之前先在数字沙盘上跑一遍——看看不同排产策略下在制品会堆多高、换型损失有多大、哪个方案综合最优。你需要的是离散事件仿真DES。我打开编辑器用Python模拟每个工件到达→加工→等待→流转的全过程统计每个时间点的在制品数量对比不同排产策略的WIP峰值和平均值。import simpydef production_line(env, name, machines, products):for p in products:with machines.request() as req:yield req# 换型时间if p.type ! last_type:yield env.timeout(setup_time)# 加工时间yield env.timeout(p.process_time)# 统计WIPwip_level len(machines.queue) machines.countenv simpy.Environment()machines simpy.Resource(env, capacity2)env.process(production_line(env, Line1, machines, product_list))env.run(until480) # 模拟一个班次分钟就这些小刘瞪大了眼睛。核心逻辑就这些。我运行了完整仿真屏幕上跳出了三种排产策略的WIP对比图排产策略 换型次数 换型损失 WIP峰值 WIP均值 综合评分────────────────────────────────────────────────────────────────────集中排产(A→B→C) 2次 90 min 142件 89件 ⚠️ WIP过高均匀穿插(ABCABC…) 15次 675 min ❌ 不可行 — ❌ 换型太多分批穿插(5A5B5C…) 8次 360 min 98件 67件 ✅ 最优你看我指着图集中排产WIP峰值142件缓存区只有120个工位直接爆仓。均匀穿插换型损失太大一个班次光换型就占了一半时间。分批穿插——每5件换一次——WIP峰值压到98件换型损失也在可接受范围内。小刘把图表截图发到排产群里下个月第三条线就按555分批穿插执行。这就是仿真告诉我们的。那条WIP曲线帮我们把拍脑袋排产变成了数据驱动的排产决策。一、实际应用场景真实痛点场景设定制造企业面临多品种小批量订单需要在同一条产线上混线生产不同产品。不同产品的工艺路线、加工时间、工装夹具不同换型Setup需要时间。生产计划部门需要回答不同排产策略集中排产 vs 均匀穿插 vs 分批穿插下在制品库存WIP压力有多大换型损失多少哪个方案综合最优现场原话叙事化我们计划部有句老话排产排得好车间没烦恼排产排不好全线乱糟糟。小刘说问题是多品种混线就像在一条高速公路上同时跑轿车、卡车和拖拉机——速度不一样换道换型要时间还不能让某一种车堵死整条路WIP爆仓。那你们不能凭经验排吗我问。凭经验小刘苦笑三种产品、每种几百件、换型45分钟——排列组合的方案有上千种。你让我凭经验选选错了要么换型浪费一天要么在制品堆到天花板要么交不了货被客户罚款。所以你要的是离散事件仿真驱动的排产评估系统——在数字沙盘上模拟不同排产策略量化WIP和换型损失选出最优方案。核心矛盾多品种混线需要平衡换型损失与WIP压力与现有MES只能记录实际生产无法预演排产方案之间的冲突。需要一个离散事件仿真与排产策略评估程序用DES模拟工件流转统计WIP对比排产策略。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述生产计划与排程 混线排产多品种共用产线需平衡效率与库存。智能制造技术基础MES、PLC 执行数据实际生产记录但缺乏预演能力。新一代支撑技术离散事件仿真DES 数字沙盘在虚拟环境中模拟生产全过程。智能工厂与智能生产柔性制造、动态调度 混线生产同一条产线灵活切换不同产品。演进范式刚性产线 → 柔性产线 → 可重构产线 从一种产品一条线到多种产品一条线用仿真优化排产。一句话总结我们需要构建一个离散事件仿真与排产策略评估程序用DES模拟多品种混线生产统计WIP对比不同排产策略的综合表现。三、核心逻辑讲解大白话3.1 问题本质把混线生产想象成高速公路收费站把多品种混线生产想象成多条车道的收费站* 每种产品 不同类型的车轿车A产品加工快、卡车B产品加工慢、大巴C产品加工最慢。* 加工设备 收费站通道只有2个通道2台机器所有车都要排队过。* 换型 换收费方式轿车用ETC快卡车用现金慢从ETC切换到现金要花时间换型时间。* 在制品WIP 收费站前排队的车如果前面的车太慢后面的车就全堵着——WIP升高。* 排产策略 放车顺序* 集中排产先放完所有轿车再放卡车最后放大巴。好处不用频繁切换收费方式换型少坏处大巴来得晚在后面排长队WIP高。* 均匀穿插轿车、卡车、大巴轮流来。好处每种车都不用等太久坏处每过一辆车就要切换一次收费方式换型爆炸。* 分批穿插每5辆轿车、5辆卡车、5辆大巴为一组轮流放。好处平衡了换型和排队。工业应用* 离散事件仿真DES用simpy库Python的DES框架模拟工件到达→请求机器→换型→加工→离开的全过程。* 在制品统计在每个时间点WIP 正在加工的工件数 等待队列中的工件数。* 排产策略通过不同的产品序列列表顺序来模拟不同排产策略。3.2 业务逻辑 → 代码映射定义产品类型│▼ Product产品1. 产品ID、类型A/B/C2. 加工时间、换型时间│▼ ProductionLine生产线DES环境1. 机器资源Resource2. 产品序列排产方案│▼ Scheduler排产器1. 集中排产AAA...BBB...CCC...2. 均匀穿插ABCABCABC...3. 分批穿插AAAAABBBBBCCCCC...│▼ WIPMonitorWIP监控器1. 每个时间步记录WIP2. 计算峰值、均值、标准差│▼ Evaluator评估器1. 换型次数和损失2. WIP指标3. 综合评分│▼ Visualizer.plot()可视化1. WIP随时间变化曲线2. 三种策略对比柱状图│▼ ReportGenerator.generate()生成报告1. 各策略指标对比2. 最优方案推荐3.3 为什么用离散事件仿真而不是Excel计算* 问题Excel可以算平均WIP但无法模拟工件A在10:03到达发现机器被工件B占用于是等待15分钟这种动态交互。* 处理策略DES是事件驱动的——只在状态变化时更新如工件到达、机器空闲能精确模拟排队、等待、资源竞争。* 工程合理性DES是工业界标准的生产仿真方法如Tecnomatix、FlexSim用simpy可以在Python中轻量级实现。3.4 排产策略的本质策略 换型次数 WIP峰值 适用场景集中排产 最少N-1次 最高 交期不急、缓存区大均匀穿插 最多N次 最低 换型时间极短5min分批穿插 中等 中等 平衡型大多数场景四、OOP 代码实现4.1 项目结构mixed_model_des/├── mixed_model_des.py # 核心代码├── test_mixed_model_des.py # 单元测试├── results/ # 输出结果│ ├── wip_curves.png # WIP随时间变化曲线│ ├── strategy_comparison.png # 策略对比柱状图│ ├── simulation_report.txt # 分析报告│ └── wip_data.csv # WIP时间序列数据└── README.md4.2 核心源码detailssummary/summary离散事件仿真多品种混线生产统计在制品库存评估不同排产策略课程映射长安大学《智能制造导论》概述生产计划与排程技术基础MES、PLC支撑技术离散事件仿真DES智能工厂柔性制造、动态调度演进范式刚性产线 → 柔性产线 → 可重构产线技术栈严格numpy # 数组运算pandas # 结果统计matplotlib # 可视化scipy # 统计检验simpy # 离散事件仿真from __future__ import annotationsimport osfrom dataclasses import dataclassfrom pathlib import Pathfrom typing import List, Dict, Tupleimport warningsimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltimport matplotlib.patches as mpatchesplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseimport simpyfrom scipy import stats# ----------------------------------------------------------------------# 1. 产品定义# ----------------------------------------------------------------------dataclassclass Product:产品类型product_id: strproduct_type: strprocess_time: float # 加工时间分钟setup_group: str # 换型组同组无需换型# ----------------------------------------------------------------------# 2. 排产器# ----------------------------------------------------------------------class Scheduler:生成不同排产策略的产品序列staticmethoddef concentrated(products: List[Product]) - List[Product]:集中排产按类型分组A全部→B全部→C全部df pd.DataFrame([{type: p.product_type, p: p}for p in products])sorted_products []for t in sorted(df[type].unique()):sorted_products.extend(df[df[type] t][p].tolist())return sorted_productsstaticmethoddef uniform_interleave(products: List[Product]) - List[Product]:均匀穿插ABCABC...循环df pd.DataFrame([{type: p.product_type, p: p}for p in products])types df[type].unique()type_lists {t: df[df[type] t][p].tolist() for t in types}result []max_len max(len(v) for v in type_lists.values())for i in range(max_len):for t in types:if i len(type_lists[t]):result.append(type_lists[t][i])return resultstaticmethoddef batch_interleave(products: List[Product],batch_size: int 5) - List[Product]:分批穿插每batch_size件一种产品df pd.DataFrame([{type: p.product_type, p: p}for p in products])types df[type].unique()type_lists {t: df[df[type] t][p].tolist() for t in types}result []max_len max(len(v) for v in type_lists.values())for i in range(0, max_len, batch_size):for t in types:batch type_lists[t][i:i batch_size]result.extend(batch)return result# ----------------------------------------------------------------------# 3. 生产线仿真离散事件# ----------------------------------------------------------------------class ProductionLine:离散事件仿真生产线def __init__(self, env: simpy.Environment,num_machines: int 2,setup_time: float 45.0):self.env envself.machines simpy.Resource(env, capacitynum_machines)self.setup_time setup_timeself.setup_group None # 当前机器设置的换型组self.wip_history [] # (时间, WIP)self.production_log []self.last_type Noneself.completed 0# 监控进程self.monitor_proc env.process(self._monitor())def _monitor(self):每1分钟记录一次WIPwhile True:yield self.env.timeout(1.0)wip len(self.machines.queue) self.machines.countself.wip_history.append((self.env.now, wip))def process_product(self, product: Product):加工单个产品的过程with self.machines.request() as req:yield req# 换型if self.setup_group ! product.setup_group:yield self.env.timeout(self.setup_time)self.setup_group product.setup_group# 加工start_time self.env.nowyield self.env.timeout(product.process_time)self.completed 1self.production_log.append({product_id: product.product_id,type: product.product_type,start: start_time,end: self.env.now,})def run_simulation(products: List[Product],num_machines: int 2,setup_time: float 45.0,sim_time: float 480.0) - ProductionLine:运行一次仿真env simpy.Environment()line ProductionLine(env, num_machines, setup_time)def product_generator():for p in products:yield env.process(line.process_product(p))env.process(product_generator())env.run(untilsim_time)return line# ----------------------------------------------------------------------# 4. 评估器# ----------------------------------------------------------------------class Evaluator:评估排产策略staticmethoddef calculate_setup_stats(products: List[Product]) - Dict:计算换型统计if not products:return {setup_count: 0, setup_time_total: 0}setup_count 0last_group Nonefor p in products:if last_group is not None and last_group ! p.setup_group:setup_count 1last_group p.setup_groupreturn {setup_count: setup_count,setup_time_total: setup_count * 45.0,}staticmethoddef calculate_wip_stats(line: ProductionLine) - Dict:计算WIP统计if not line.wip_history:return {wip_peak: 0, wip_mean: 0, wip_std: 0}wip_values [w for _, w in line.wip_history]return {wip_peak: max(wip_values),wip_mean: np.mean(wip_values),wip_std: np.std(wip_values),wip_values: wip_values,times: [t for t, _ in line.wip_history],}# ----------------------------------------------------------------------# 5. 可视化器# ----------------------------------------------------------------------class Visualizer:可视化分析结果def __init__(self):self.results_dir Path(results)os.makedirs(self.results_dir, exist_okTrue)def plot_wip_curves(self, results: Dict[str, Dict]):绘制各策略的WIP曲线print([INFO] 绘制WIP曲线...)fig, ax plt.subplots(figsize(14, 6))colors {集中排产: #E74C3C, 均匀穿插: #3498DB,分批穿插: #27AE60}for name, data in results.items():wip_stats data[wip_stats]if times in wip_stats and wip_values in wip_stats:ax.plot(wip_stats[times], wip_stats[wip_values],labelname, colorcolors.get(name, #999999),linewidth2, alpha0.8)ax.set_xlabel(仿真时间 (分钟), fontsize12)ax.set_ylabel(在制品数量 (件), fontsize12)ax.set_title(不同排产策略的在制品库存变化,fontsize14, fontweightbold)ax.legend(fontsize11)ax.grid(True, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / wip_curves.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / wip_curves.png})def plot_strategy_comparison(self, results: Dict[str, Dict]):绘制策略对比柱状图print([INFO] 绘制策略对比图...)fig, axes plt.subplots(2, 2, figsize(14, 10))names list(results.keys())x np.arange(len(names))w 0.25# WIP峰值wip_peaks [results[n][wip_stats][wip_peak] for n in names]axes[0, 0].bar(x, wip_peaks, color[#E74C3C, #3498DB, #27AE60],alpha0.8)axes[0, 0].set_ylabel(WIP峰值 (件), fontsize12)axes[0, 0].set_title(WIP峰值对比, fontsize13, fontweightbold)axes[0, 0].set_xticks(x)axes[0, 0].set_xticklabels(names, rotation10)axes[0, 0].grid(True, alpha0.3, axisy)# WIP均值wip_means [results[n][wip_stats][wip_mean] for n in names]axes[0, 1].bar(x, wip_means, color[#E74C3C, #3498DB, #27AE60],alpha0.8)axes[0, 1].set_ylabel(WIP均值 (件), fontsize12)axes[0, 1].set_title(WIP均值对比, fontsize13, fontweightbold)axes[0, 1].set_xticks(x)axes[0, 1].set_xticklabels(names, rotation10)axes[0, 1].grid(True, alpha0.3, axisy)# 换型次数setup_counts [results[n][setup_stats][setup_count] for n in names]axes[1, 0].bar(x, setup_counts,color[#E74C3C, #3498DB, #27AE60], alpha0.8)axes[1, 0].set_ylabel(换型次数, fontsize12)axes[1, 0].set_title(换型次数对比, fontsize13, fontweightbold)axes[1, 0].set_xticks(x)axes[1, 0].set_xticklabels(names, rotation10)axes[1, 0].grid(True, alpha0.3, axisy)# 换型损失时间setup_times [results[n][setup_stats][setup_time_total] / 60for n in names]axes[1, 1].bar(x, setup_times,color[#E74C3C, #3498DB, #27AE60], alpha0.8)axes[1, 1].set_ylabel(换型损失 (小时), fontsize12)axes[1, 1].set_title(换型损失时间对比, fontsize13, fontweightbold)axes[1, 1].set_xticks(x)axes[1, 1].set_xticklabels(names, rotation10)axes[1, 1].grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / strategy_comparison.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / strategy_comparison.png})# ----------------------------------------------------------------------# 6. 报告生成器# ----------------------------------------------------------------------class ReportGenerator:分析报告生成器def __init__(self):self.results_dir Path(results)os.makedirs(self.results_dir, exist_okTrue)def generate(self, results: Dict[str, Dict]) - str:生成报告print([INFO] 生成分析报告...)report_lines []report_lines.append( * 80)report_lines.append(多品种混线生产排产策略评估报告)report_lines.append( * 80)report_lines.append(f\n{策略:12} {换型次数:10} {换型损失:12} f{WIP峰值:10} {WIP均值:10})report_lines.append(- * 60)for name, data in results.items():setup data[setup_stats]wip data[wip_stats]report_lines.append(f{name:12} {setup[setup_count]:10} f{setup[setup_time_total]/60:10.1f}h f{wip[wip_peak]:10.0f} f{wip[wip_mean]:10.1f})# 综合评分简单加权report_lines.append(f\n综合建议:)report_lines.append(- * 40)best min(results.keys(),keylambda n: results[n][wip_stats][wip_peak])report_lines.append(f 推荐策略: {best})report_lines.append(f 理由: WIP峰值最低综合平衡换型损失)report_lines.append(\n * 80)report_lines.append(报告生成完毕)report_lines.append( * 80)report_text \n.join(report_lines)report_path self.results_dir / simulation_report.txtwith open(report_path, w, encodingutf-8) as f:f.write(report_text)print(f 报告已保存: {report_path})return report_text# ----------------------------------------------------------------------# 7. 主程序演示# ----------------------------------------------------------------------def demo():完整演示流程print( * 80)print(离散事件仿真多品种混线生产评估排产策略WIP压力)print( * 80)# 1. 创建产品列表print(\n[INFO] 步骤1: 创建产品列表...)products []# A产品 500件for i in range(500):products.append(Product(fA{i}, A, 5.0, A))# B产品 300件for i in range(300):products.append(Product(fB{i}, B, 8.0, B))# C产品 200件for i in range(200):products.append(Product(fC{i}, C, 12.0, C))print(f 总产品数: {len(products)})# 2. 生成排产方案print(\n[INFO] 步骤2: 生成排产方案...)scheduler Scheduler()strategies {集中排产: scheduler.concentrated(products),均匀穿插: scheduler.uniform_interleave(products),分批穿插: scheduler.batch_interleave(products, batch_size5),}for name, seq in strategies.items():setup Evaluator.calculate_setup_stats(seq)print(f {name}: {len(seq)}件, 换型{setup[setup_count]}次)# 3. 运行仿真print(\n[INFO] 步骤3: 运行离散事件仿真...)results {}for name, seq in strategies.items():print(f 运行 {name}...)line run_simulation(seq, num_machines2, setup_time45.0,sim_time480.0)setup_stats Evaluator.calculate_setup_stats(seq)wip_stats Evaluator.calculate_wip_stats(line)results[name] {setup_stats: setup_stats,wip_stats: wip_stats,}print(f WIP峰值: {wip_stats[wip_peak]:.0f}, f均值: {wip_stats[wip_mean]:.1f})# 4. 可视化print(\n[INFO] 步骤4: 可视化...)vis Visualizer()vis.plot_wip_curves(results)vis.plot_strategy_comparison(results)# 5. 生成报告print(\n[INFO] 步骤5: 生成报告...)report_gen ReportGenerator()report_text report_gen.generate(results)# 保存WIP数据for name, data in results.items():if times in data[wip_stats]:wip_df pd.DataFrame({time: data[wip_stats][times],wip: data[wip_stats][wip_values],})wip_df.to_csv(fresults/wip_{name}.csv, indexFalse)# 摘要print(\n * 80)print(分析报告摘要)print( * 80)print(report_text[:1200] \n... if len(report_text) 1200 else report_text)print(\n 工程落地建议)print( 1. 接入MES获取实际加工时间和换型时间)print( 2. 增加设备数量、故障率等参数)print( 3. 结合订单交期约束进行多目标优化)return resultsif __name__ __main__:demo()/detailsdetailssummary/summaryimport osimport pytestimport numpy as npimport pandas as pdfrom pathlib import Pathfrom mixed_model_des import (Product, Scheduler, ProductionLine, run_simulation,Evaluator, Visualizer, ReportGenerator)pytest.fixturedef sample_products():创建测试产品列表products []for i in range(10):products.append(Product(fA{i}, A, 5.0, A))for i in range(5):products.append(Product(fB{i}, B, 8.0, B))return productsdef test_scheduler_concentrated(sample_products):seq Scheduler.concentrated(sample_products)assert len(seq) len(sample_products)# 集中排产所有A在前types [p.product_type for p in seq]assert all(t A for t in types[:10])def test_scheduler_uniform_interleave(sample_products):seq Scheduler.uniform_interleave(sample_products)assert len(seq) len(sample_products)利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛