工单-质检-能耗三表关联用Python算出每批产品的能耗账和质量账周三上午生产部的小李拿着三张Excel表一脸愁容地走进办公室。哥帮个忙。小李把三张表摊在桌上厂长昨天开会问我咱们不同工单生产的产品能耗差别大吗不良率跟能耗有没有关系我看着这三张表头都大了。我扫了一眼第一张是工单表记录了工单号、产品型号、计划数量、实际产出第二张是质检表记录了工单号、检测数量、不良数量、不良类型第三张是能耗表记录了工单号、总耗电量kWh。这不就是三张表关联一下算两个指标吗我说。你以为简单小李苦笑工单表有156行质检表有203行——因为一个工单可能分多次质检能耗表有148行——有些工单还没录能耗。我想把三张表按工单号对齐算每个工单的单位产品能耗和不良率结果VLOOKUP一拉要么对不上要么重复计算。这就是多表关联的问题。我打开编辑器用 pandas 的merge三行代码搞定。import pandas as pd# 1. 读取三张表orders pd.read_csv(orders.csv)quality pd.read_csv(quality.csv)energy pd.read_csv(energy.csv)# 2. 质检表按工单聚合一个工单多次质检求和q_agg quality.groupby(order_id).agg(total_inspected(inspected, sum),total_defects(defects, sum)).reset_index()q_agg[defect_rate] q_agg[total_defects] / q_agg[total_inspected]# 3. 三表关联merged orders.merge(q_agg, onorder_id, howinner)merged merged.merge(energy[[order_id, energy_kwh]], onorder_id, howinner)# 4. 计算单位产品能耗merged[energy_per_unit] merged[energy_kwh] / merged[actual_output]print(merged[[order_id, energy_per_unit, defect_rate]].head())就这些小李瞪大了眼睛。核心逻辑就这些。我运行了一下屏幕上跳出了结果order_id energy_per_unit defect_rate0 WO-1001 2.34 0.01231 WO-1002 3.87 0.03412 WO-1003 2.18 0.0089你看我指着屏幕WO-1002的单位能耗是WO-1003的1.77倍不良率也高出近3倍。这说明能耗高的工单质量也可能有问题——可能是设备参数没调好空转时间长既费电又出废品。小李把结果截图发给了厂长附了一句以前我只知道电用了多少、废品有多少现在我知道每批产品的能耗和质量到底什么关系。这一次三表关联帮我们看见了看不见的工单画像。一、实际应用场景真实痛点场景设定制造企业的生产管理涉及多张数据表——工单表记录生产任务和产出质检表记录检测结果能耗表记录各工单的电耗。生产管理人员需要将三张表关联计算每个工单的单位产品能耗kWh/件和不良率%以评估不同工单的生产效率和质量水平发现高能耗、高不良率的异常工单为工艺优化提供数据支撑。面对不同粒度的数据和一对多的关联关系手工处理极易出错。现场原话叙事化我们车间有句老话电费是老板的废品是自己的。小李说但到底哪些工单既费电又出废品以前没人算过。因为数据散在三张表里工单表在MES里质检表在QMS里能耗表在能源管理系统里。每次要分析得分别导出再手工拼。那你们的BI系统不能自动关联吗我问。有BI但IT说要开发报表得走流程排期到下个月。小李摊手厂长明天就要看结果我等不了。所以你要的是离线多表关联分析——把三张表按工单号对齐算单位能耗和不良率再找异常。对。而且不只是算出来。小李补充我还想看趋势——不同产品型号的单位能耗是不是不一样同一型号不同批次的不良率波动大不大能耗和不良率之间有没有相关性如果高能耗的工单不良率也高那说明问题出在设备参数上而不是原材料。核心矛盾生产管理需要量化每批产品的能耗效率和质量水平以优化工艺与多源数据分散在不同系统中缺乏快速关联与综合分析能力之间的冲突。需要一个三表关联分析程序自动对齐工单、质检、能耗数据计算关键指标并挖掘关联规律。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述生产管理、制造执行 工单管理工单是生产执行的基本单元关联产出、质量、能耗数据。智能制造技术基础制造过程数据采集 数据关联工单、质检、能耗数据来自不同环节需要关联分析。新一代支撑技术工业大数据多源数据融合 数据融合将分散的工单、质量、能耗数据按工单号关联形成统一视图。智能工厂与智能生产能效管理、质量管理 综合指标单位产品能耗和不良率是衡量生产效率和质量的核心KPI。演进范式手工台账 → 单表统计 → 多表关联分析 → 实时综合看板 从三张表各看各的到一次关联看清全局实现生产数据的深度融合。一句话总结我们需要构建一个工单-质检-能耗三表关联分析程序通过多表关联计算每个工单的单位产品能耗和不良率评估生产效率和质量水平发现异常工单。三、核心逻辑讲解大白话3.1 问题本质把三表关联看成拼拼图把三表关联和拼拼图的关系想象成把三块碎片拼成一幅完整的画* 工单表 拼图底板每个工单是一块底板上面有工单号、产品型号、计划数量、实际产出。* 质检表 碎片A记录每个工单的检测结果。注意一个工单可能检测了多次不同班次、不同批次所以质检表中有多个行对应同一个工单号——这是一对多关系。* 能耗表 碎片B记录每个工单的耗电量。通常一个工单一条记录但也可能因为数据采集方式而有重复或缺失。* 关联键 拼图的卡口三张表都有order_id工单号这就是把它们拼在一起的卡口。* 聚合 把碎片A压扁质检表一对多需要按工单号分组把多次检测的数量和不良数加起来变成一对一。* merge 拼合用pd.merge() 把三张表按工单号拼在一起形成一张大宽表。* 计算指标 在拼好的图上写字用产出数量算出单位能耗用检测数量算出不良率。工业应用* 输入三张CSV——orders.csv工单表、quality.csv质检表、energy.csv能耗表。* 聚合quality.groupby(order_id).sum() 合并同一工单的多次质检。* 关联orders.merge(q_agg, onorder_id) 左连接工单和质检merge(energy, onorder_id) 再关联能耗。* 计算energy_kwh / actual_output 单位产品能耗total_defects / total_inspected 不良率。* 分析按产品型号分组看平均能耗和不良率用散点图看能耗与不良率的相关性。3.2 业务逻辑 → 代码映射读取三张表│▼ DataLoader.load()数据加载1. pd.read_csv() 读取工单、质检、能耗表2. 数据清洗去重、缺失值处理│▼ QualityAggregator.aggregate()质检数据聚合1. groupby(order_id) 按工单汇总检测数和不良数2. 计算不良率│▼ DataMerger.merge()三表关联1. orders.merge(q_agg) 工单 质检2. merged.merge(energy) 能耗3. 处理未匹配的记录外连接标记│▼ MetricsCalculator.calculate()指标计算1. 单位产品能耗 总能耗 / 实际产出2. 不良率 不良数 / 检测数3. 按产品型号分组统计│▼ Visualizer.plot()可视化1. 散点图能耗 vs 不良率看相关性2. 柱状图各产品型号的平均单位能耗3. 柱状图各产品型号的平均不良率4. 箱线图各型号能耗分布│▼ ReportGenerator.generate_report()生成报告1. 异常工单标记高能耗或高不良率2. 综合排名3. 优化建议3.3 为什么用merge 而不是直接用Excel VLOOKUP*mergepandas 的向量化关联操作支持内连接、左连接、右连接、外连接。一行代码完成关联自动处理重复键速度快可复现。* VLOOKUPExcel 的单列查找遇到一对多关系会只返回第一条匹配需要手动处理重复值。大数据量时卡顿且公式易错。* 工程选择本例用merge 做多表关联配合groupby 处理一对多关系确保数据完整性。3.4 如何处理质检表一对多的问题* 问题一个工单可能分多次质检如每班检测一次质检表中同一工单号出现多次。直接关联会导致工单表的行被复制多次笛卡尔膨胀。* 处理策略在关联之前先对质检表按工单号分组聚合把多次检测的数量和不良数求和得到每个工单的累计检测数和累计不良数再计算不良率。这样质检表就变成了一对一的关系。* 本例处理在QualityAggregator 中执行groupby(order_id).agg({inspected: sum, defects: sum})然后再关联。四、OOP 代码实现4.1 项目结构order_quality_energy/├── data/│ ├── orders.csv # 工单表│ ├── quality.csv # 质检表│ └── energy.csv # 能耗表├── results/ # 输出结果│ ├── merged_data.csv # 关联后的大宽表│ ├── scatter_energy_defect.png # 能耗-不良率散点图│ ├── bar_energy_by_model.png # 各型号单位能耗│ ├── bar_defect_by_model.png # 各型号不良率│ ├── box_energy_by_model.png # 各型号能耗箱线图│ └── analysis_report.txt # 分析报告├── order_quality_energy.py # 核心代码├── test_order_quality_energy.py # 单元测试├── README.md└── requirements.txt4.2 核心源码detailssummary/summary工单-质检-能耗三表关联分析统计不同工单的单位产品能耗与不良率课程映射长安大学《智能制造导论》概述生产管理、制造执行技术基础制造过程数据采集支撑技术工业大数据多源数据融合智能工厂能效管理、质量管理演进范式手工台账 → 单表统计 → 多表关联分析 → 实时综合看板技术栈严格numpy # 数值计算pandas # 多表关联、聚合matplotlib # 可视化networkx # 无scikit-learn # 无scipy # 无torch # 无from __future__ import annotationsimport osfrom dataclasses import dataclass, fieldfrom pathlib import Pathfrom typing import List, Optional, Dictimport numpy as npimport pandas as pdimport matplotlib.pyplot as pltplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] False# ----------------------------------------------------------------------# 1. 配置# ----------------------------------------------------------------------dataclassclass AnalysisConfig:分析配置data_dir: str dataresults_dir: str results# 文件名orders_file: str orders.csvquality_file: str quality.csvenergy_file: str energy.csv# 列名order_id_col: str order_idmodel_col: str product_modelplanned_col: str planned_qtyactual_col: str actual_outputinspected_col: str inspected_qtydefects_col: str defectsenergy_col: str energy_kwh# 异常阈值energy_anomaly_threshold: float 5.0 # 单位能耗超过5 kWh/件defect_anomaly_threshold: float 0.05 # 不良率超过5%random_seed: int 42# ----------------------------------------------------------------------# 2. 数据加载器# ----------------------------------------------------------------------class DataLoader:数据加载器def __init__(self, config: AnalysisConfig):self.config configself.data_dir Path(config.data_dir)os.makedirs(self.data_dir, exist_okTrue)def generate_synthetic_data(self, n_orders: int 50):生成模拟数据print(f[INFO] 生成模拟数据{n_orders}个工单...)np.random.seed(self.config.random_seed)# 产品型号models [M-A100, M-B200, M-C300, M-D400]# 工单表orders []for i in range(1, n_orders 1):model np.random.choice(models)planned np.random.randint(500, 2000)# 实际产出略低于计划有损耗actual int(planned * np.random.uniform(0.92, 1.0))orders.append({order_id: fWO-{1000 i},product_model: model,planned_qty: planned,actual_output: actual})# 质检表一个工单1-3次质检quality []for order in orders:n_inspections np.random.randint(1, 4)remaining order[actual_output]for j in range(n_inspections):if j n_inspections - 1:inspected remainingelse:inspected np.random.randint(remaining // 3, remaining // 2)remaining - inspected# 不良率基础值 随机波动base_defect {M-A100: 0.01, M-B200: 0.02, M-C300: 0.015, M-D400: 0.03}defect_rate base_defect[order[product_model]] * np.random.uniform(0.5, 2.0)defects int(inspected * defect_rate)quality.append({order_id: order[order_id],inspected_qty: inspected,defects: defects})# 能耗表每个工单一条偶尔缺失energy []for order in orders:if np.random.random() 0.05: # 5%概率缺失continue# 单位能耗基础值 随机波动base_energy {M-A100: 2.0, M-B200: 3.0, M-C300: 2.5, M-D400: 4.0}energy_per_unit base_energy[order[product_model]] * np.random.uniform(0.8, 1.4)total_energy energy_per_unit * order[actual_output]energy.append({order_id: order[order_id],energy_kwh: round(total_energy, 2)})# 保存self.data_dir.mkdir(parentsTrue, exist_okTrue)pd.DataFrame(orders).to_csv(self.data_dir / self.config.orders_file, indexFalse)pd.DataFrame(quality).to_csv(self.data_dir / self.config.quality_file, indexFalse)pd.DataFrame(energy).to_csv(self.data_dir / self.config.energy_file, indexFalse)print(f 工单: {len(orders)})print(f 质检记录: {len(quality)})print(f 能耗记录: {len(energy)})return pd.DataFrame(orders), pd.DataFrame(quality), pd.DataFrame(energy)def load_data(self) - tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:加载三张表print(f[INFO] 加载数据...)orders_path Path(self.config.data_dir) / self.config.orders_filequality_path Path(self.config.data_dir) / self.config.quality_fileenergy_path Path(self.config.data_dir) / self.config.energy_fileif not (orders_path.exists() and quality_path.exists() and energy_path.exists()):self.generate_synthetic_data()orders_df pd.read_csv(orders_path)quality_df pd.read_csv(quality_path)energy_df pd.read_csv(energy_path)print(f 工单: {len(orders_df)} 条)print(f 质检: {len(quality_df)} 条)print(f 能耗: {len(energy_df)} 条)return orders_df, quality_df, energy_df# ----------------------------------------------------------------------# 3. 质检数据聚合器# ----------------------------------------------------------------------class QualityAggregator:聚合质检数据一对多 → 一对一def __init__(self, config: AnalysisConfig):self.config configdef aggregate(self, quality_df: pd.DataFrame) - pd.DataFrame:按工单号聚合质检数据print(f[INFO] 聚合质检数据...)q_agg quality_df.groupby(self.config.order_id_col).agg(total_inspected(self.config.inspected_col, sum),total_defects(self.config.defects_col, sum)).reset_index()# 计算不良率q_agg[defect_rate] q_agg[total_defects] / q_agg[total_inspected]print(f 聚合后: {len(q_agg)} 个工单)return q_agg# ----------------------------------------------------------------------# 4. 数据关联器# ----------------------------------------------------------------------class DataMerger:三表关联def __init__(self, config: AnalysisConfig):self.config configdef merge(self, orders_df: pd.DataFrame, q_agg: pd.DataFrame,energy_df: pd.DataFrame) - pd.DataFrame:关联三张表print(f[INFO] 关联三张表...)# 工单 质检左连接保留所有工单merged orders_df.merge(q_agg,onself.config.order_id_col,howleft)# 能耗内连接只保留有能耗数据的工单merged merged.merge(energy_df[[self.config.order_id_col, self.config.energy_col]],onself.config.order_id_col,howinner)# 计算单位产品能耗merged[energy_per_unit] merged[self.config.energy_col] / merged[self.config.actual_col]print(f 关联后: {len(merged)} 条记录)return merged# ----------------------------------------------------------------------# 5. 指标计算器# ----------------------------------------------------------------------class MetricsCalculator:计算分析指标def __init__(self, config: AnalysisConfig):self.config configdef calculate_by_model(self, merged: pd.DataFrame) - pd.DataFrame:按产品型号分组统计print(f[INFO] 按产品型号统计...)model_stats merged.groupby(self.config.model_col).agg(order_count(self.config.order_id_col, count),avg_energy_per_unit(energy_per_unit, mean),avg_defect_rate(defect_rate, mean),std_energy(energy_per_unit, std),std_defect(defect_rate, std),total_energy(self.config.energy_col, sum),total_output(self.config.actual_col, sum)).reset_index()model_stats[overall_energy_per_unit] model_stats[total_energy] / model_stats[total_output]print(f 产品型号数: {len(model_stats)})return model_statsdef find_anomalies(self, merged: pd.DataFrame) - pd.DataFrame:标记异常工单merged merged.copy()merged[energy_anomaly] merged[energy_per_unit] self.config.energy_anomaly_thresholdmerged[defect_anomaly] merged[defect_rate] self.config.defect_anomaly_thresholdmerged[is_anomaly] merged[energy_anomaly] | merged[defect_anomaly]n_anomaly merged[is_anomaly].sum()print(f 异常工单: {n_anomaly} 个)return merged# ----------------------------------------------------------------------# 6. 可视化器# ----------------------------------------------------------------------class Visualizer:可视化分析结果def __init__(self, config: AnalysisConfig):self.config configself.results_dir Path(config.results_dir)os.makedirs(self.results_dir, exist_okTrue)def plot_scatter(self, merged: pd.DataFrame):能耗 vs 不良率散点图print(f[INFO] 绘制能耗-不良率散点图...)fig, ax plt.subplots(figsize(10, 7))colors {M-A100: #E74C3C, M-B200: #3498DB,M-C300: #F39C12, M-D400: #27AE60}for model in merged[self.config.model_col].unique():subset merged[merged[self.config.model_col] model]ax.scatter(subset[energy_per_unit],subset[defect_rate] * 100, # 转百分比ccolors.get(model, #999999),labelmodel,s80,alpha0.7,edgecolorswhite,linewidth0.5)# 异常阈值线ax.axvline(self.config.energy_anomaly_threshold, colorred,linestyle--, alpha0.5, label能耗阈值)ax.axhline(self.config.defect_anomaly_threshold * 100, colorred,linestyle--, alpha0.5, label不良率阈值)ax.set_xlabel(单位产品能耗 (kWh/件), fontsize12)ax.set_ylabel(不良率 (%), fontsize12)ax.set_title(单位产品能耗 vs 不良率, fontsize14, fontweightbold)ax.legend(fontsize10)ax.grid(True, alpha0.3)plt.tight_layout()plt.savefig(self.results_dir / scatter_energy_defect.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / scatter_energy_defect.png})def plot_bar_energy_by_model(self, model_stats: pd.DataFrame):各型号平均单位能耗柱状图print(f[INFO] 绘制各型号单位能耗柱状图...)fig, ax plt.subplots(figsize(10, 6))models model_stats[self.config.model_col]energies model_stats[avg_energy_per_unit]bars ax.bar(models, energies, color[#E74C3C, #3498DB, #F39C12, #27AE60],edgecolorwhite, alpha0.8)# 标注数值for bar, val in zip(bars, energies):ax.text(bar.get_x() bar.get_width() / 2, bar.get_height() 0.05,f{val:.2f}, hacenter, vabottom, fontsize10, fontweightbold)ax.set_ylabel(平均单位能耗 (kWh/件), fontsize12)ax.set_title(各产品型号平均单位产品能耗, fontsize14, fontweightbold)ax.grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / bar_energy_by_model.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / bar_energy_by_model.png})def plot_bar_defect_by_model(self, model_stats: pd.DataFrame):各型号平均不良率柱状图print(f[INFO] 绘制各型号不良率柱状图...)fig, ax plt.subplots(figsize(10, 6))models model_stats[self.config.model_col]defect_rates model_stats[avg_defect_rate] * 100 # 转百分比bars ax.bar(models, defect_rates, color[#9B59B6, #E67E22, #1ABC9C, #34495E],edgecolorwhite, alpha0.8)for bar, val in zip(bars, defect_rates):ax.text(bar.get_x() bar.get_width() / 2, bar.get_height() 0.1,f{val:.2f}%, hacenter, vabottom, fontsize10, fontweightbold)ax.set_ylabel(平均不良率 (%), fontsize12)ax.set_title(各产品型号平均不良率, fontsize14, fontweightbold)ax.grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / bar_defect_by_model.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / bar_defect_by_model.png})def plot_box_energy_by_model(self, merged: pd.DataFrame):各型号能耗分布箱线图print(f[INFO] 绘制各型号能耗箱线图...)fig, ax plt.subplots(figsize(10, 6))data [merged[merged[self.config.model_col] m][energy_per_unit]for m in merged[self.config.model_col].unique()]bp ax.boxplot(data, labelsmerged[self.config.model_col].unique(),patch_artistTrue)colors [#E74C3C, #3498DB, #F39C12, #27AE60]for patch, color in zip(bp[boxes], colors):patch.set_facecolor(color)patch.set_alpha(0.8)ax.set_ylabel(单位产品能耗 (kWh/件), fontsize12)ax.set_title(各产品型号单位能耗分布, fontsize14, fontweightbold)ax.grid(True, alpha0.3, axisy)plt.tight_layout()plt.savefig(self.results_dir / box_energy_by_model.png,dpi150, bbox_inchestight)plt.close()print(f 已保存: {self.results_dir / box_energy_by_model.png})# ----------------------------------------------------------------------# 7. 报告生成器# ----------------------------------------------------------------------class ReportGenerator:分析报告生成器def __init__(self, config: AnalysisConfig):self.config configself.results_dir Path(config.results_dir)os.makedirs(self.results_dir, exist_okTrue)def generate(self, merged: pd.DataFrame, model_stats: pd.DataFrame) - str:生成报告print(f[INFO] 生成分析报告...)report_lines []report_lines.append( * 80)report_lines.append(工单-质检-能耗关联分析报告)report_lines.append( * 80)# 概况report_lines.append(f\n数据概况:)利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
