周四下午，FMS 车间调度室。
"这个月第三批订单又延期了。"生产主管老刘把工单 Excel 表甩到桌上，转头冲着调度系统喊，"小王，你那个排产软件不是说能算出来吗？结果还是拖了两天！"
我拉过椅子坐下，打开那份工单表。几百行数据，字段包括：工单号、产品类型（A型法兰/B型壳体/C型轴套）、计划加工时长（分钟）、实际加工时长（分钟）、设备编号、操作员、完成状态。问题一眼就看到了——
"小王，你排产的时候，A 型法兰的计划时长是按什么给的？"
"按经验啊，"小王挠挠头，"之前做的几批，差不多都是 90 分钟一个。"
"差不多？"我指着屏幕，"你看这组数据——A 型法兰实际加工时长从 72 分钟到 156 分钟，波动快一倍了。你用 90 分钟去排产，遇到实际要 150 分钟的批次，整条线就堵死了。"
老刘凑过来："那到底该用多少分钟排产？"
"得按产品类型分组，算每类工单的实际平均加工时长，"我说，"不是拍脑袋'差不多'，是用历史数据说话。"
我敲了几行代码：
import pandas as pd
import matplotlib.pyplot as plt
# 加载工单数据
df = pd.read_excel("fms_workorders.xlsx")
# 按产品类型分组，计算平均实际加工时长
avg_times = df.groupby("product_type")["actual_time"].mean()
# 画柱状图
avg_times.plot(kind="bar", color="#3498DB")
plt.ylabel("平均加工时长 (分钟)")
plt.title("各类产品平均加工时长")
plt.show()
print(avg_times)
# A型法兰 112.3
# B型壳体 87.6
# C型轴套 54.2
"这只是核心逻辑，"我说，"完整版会用 OOP 封装：一个类管工单数据加载，一个类管数据清洗（处理空值、异常值），一个类管分组统计（按产品类型算平均/最大/最小加工时长），一个类管可视化（柱状图+箱线图）。数据自包含——用 numpy 合成一份工单 CSV，读者下载就能跑。"
老刘盯着屏幕："所以你这东西，就是把几百个工单按产品类型分堆，每堆算一个靠谱的平均加工时间——以后排产就用这个数，不用再'差不多'了。"
"对。而且不光是平均，"我补充，"箱线图还能看出每类产品的加工时长波动范围——波动大的产品，排产时要多留缓冲时间。"
小王点点头："那我明天就把这个数嵌进排产逻辑里。"
一、实际应用场景（真实痛点）
场景设定：柔性制造系统（FMS）车间承接多品种小批量订单，每种产品（A型法兰、B型壳体、C型轴套等）的加工工艺路径相似但参数不同，历史工单记录了计划加工时长和实际加工时长。现场常见痛点：排产时凭经验估算加工时长，未利用历史工单数据计算各类产品的实际平均加工时长，导致排产偏差大、订单延期、设备利用率低。
现场原话（叙事化）：
"我不是没有数据，"小王说，"工单系统里几百条历史记录，计划时间和实际时间都有。但我从来没认真分析过——每次排产都是凭感觉给个数，差不多就行。"
"差不多就是差很多，"我说，"A 型法兰你给 90 分钟，实际平均要 112 分钟，每批 10 个就是多出来将近 4 个小时。一天排三批，直接拖到第二天。"
核心矛盾："经验估算的加工时长"与"基于历史数据的实际平均加工时长"之间的偏差。需要一个"FMS 工单加工时长分析程序"，用
"pandas" 加载工单数据，按产品类型分组计算平均/统计加工时长，用
"matplotlib" 可视化对比，为排产提供数据支撑。
二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）
《先进制造技术》模块 本篇痛点对应
柔性制造系统FMS与先进生产管理：生产计划与排程 工单加工时长统计：排产的基础输入是准确的工序时间，经验估算导致计划偏差。
先进制造技术基础：制造系统性能评估 加工效率分析：通过历史数据统计各类产品的实际加工时长，评估制造系统效率。
智能制造与数字孪生：制造大数据驱动决策 数据驱动的排产优化：从工单历史数据中挖掘规律，替代经验判断。
数控加工与CAD/CAM技术：数控加工工艺规划 工艺参数与加工时间关联：不同产品的数控加工程序复杂度不同，加工时长存在差异。
一句话总结：我们需要构建一个"FMS 工单加工时长分析程序"，用
"pandas" 加载和分组统计工单数据，用
"numpy" 计算统计指标，用
"matplotlib" 绘制柱状图和箱线图，输出各类产品的平均加工时长及波动范围，为排产决策提供数据依据。
三、核心逻辑讲解（大白话）
3.1 问题本质：把工单分析想象成"快递分拣计时"
把 FMS 车间的工单数据想象成"快递分拣中心的历史记录"：
* 工单数据表 = 快递分拣记录：每次分拣一批包裹，记录包裹类型（小件/中件/大件）、计划分拣时间、实际分拣时间。
* 产品类型分组 = 按包裹大小分堆：小件放一堆、中件放一堆、大件放一堆。
* 计算平均加工时长 = 算每堆的平均分拣时间：小件平均 5 分钟、中件平均 12 分钟、大件平均 25 分钟。
* 排产用平均时长 = 按平均时间预估：下次来 100 个小件，按 5×100=500 分钟排计划，而不是凭感觉说"大概两小时"。
* 箱线图 = 看波动范围：小件大部分在 4~6 分钟，偶尔有 10 分钟的（异常），排产时多留点余量。
工业应用：
* pandas 加载：
"read_csv()" /
"read_excel()" 读取工单表。
* 分组统计：
"groupby("product_type")["actual_time"].agg(["mean", "std", "min", "max"])" 一行代码算出每类产品的平均、标准差、最小、最大加工时长。
* 异常值处理：用
"numpy" 结合箱线图法则（IQR）识别异常工单（如设备故障导致的异常长加工时间），避免拉偏平均值。
* matplotlib 可视化：柱状图对比各类产品平均时长，箱线图展示每类产品的时长分布和波动。
3.2 业务逻辑 → 代码映射
定义工单数据模型
│
▼ WorkorderLoader (pandas)
加载工单 CSV/Excel：
pd.read_csv("fms_workorders.csv")
解析产品类型、计划时长、实际时长
│
▼ DataCleaner (pandas/numpy)
数据清洗：
处理空值（删除或标记）
异常值检测（IQR 法则）
过滤未完成工单
│
▼ DurationAnalyzer (pandas/numpy)
加工时长分析：
groupby("product_type").agg(
mean=("actual_time", "mean"),
std=("actual_time", "std"),
min=("actual_time", "min"),
max=("actual_time", "max"),
count=("actual_time", "count"),
)
计算计划vs实际偏差
│
▼ Visualizer (matplotlib)
可视化：
1. 柱状图：各类产品平均加工时长
2. 箱线图：每类产品的时长分布
3. 对比图：计划 vs 实际（按产品类型）
│
▼ SyntheticDataGenerator (numpy)
合成数据生成：
用 numpy 按不同分布生成各类产品的加工时长
包含正常波动和少量异常值
确保数据自包含、可复现
3.3 为什么用分组统计而不是逐个看？
* 问题：几百条工单逐行翻看，无法形成"这类产品一般要多久"的整体认知。
* 处理策略：
"groupby" 按产品类型分堆，一次性算出每堆的统计指标，效率高且信息密度大。
* 工程合理性：FMS 车间产品种类可能多达几十种，分组统计是生产数据分析的标准方法。
3.4 分析前后对比
维度 经验估算 数据驱动分析
加工时长 "差不多 90 分钟" A型法兰 112.3±18.5 分钟
排产准确性 偏差 20%~50% 偏差 <10%
异常处理 不知道有异常 IQR 识别异常工单，排除后重算
可视化 无 柱状图+箱线图，一目了然
四、OOP 代码实现
4.1 项目结构
fms_workorder_analysis/
├── fms_workorder_analysis/
│ ├── __init__.py
│ ├── workorder_loader.py # 工单数据加载
│ ├── data_cleaner.py # 数据清洗与异常检测
│ ├── duration_analyzer.py # 加工时长分析
│ ├── visualizer.py # 可视化
│ └── synthetic_data.py # 合成数据生成
├── tests/
│ ├── __init__.py
│ └── test_workorder_analysis.py # 单元测试
├── results/
│ ├── avg_duration_bar.png # 平均加工时长柱状图
│ ├── duration_boxplot.png # 加工时长箱线图
│ ├── plan_vs_actual.png # 计划vs实际对比图
│ ├── duration_summary.csv # 加工时长汇总
│ └── analysis_report.txt # 分析报告
└── run_analysis.py # 主程序入口
4.2 核心源码
<details>
<summary></summary>
"""FMS 工单数据加载器"""
import pandas as pd
from pathlib import Path
from typing import Optional
class WorkorderLoader:
"""
FMS 车间工单数据加载器
支持 CSV 和 Excel 格式的工单数据。
"""
def __init__(self, filepath: str, encoding: str = "utf-8"):
self.filepath = Path(filepath)
self.encoding = encoding
self._raw_df: Optional[pd.DataFrame] = None
def load(self,
product_col: str = "product_type",
plan_col: str = "plan_time",
actual_col: str = "actual_time",
status_col: str = "status") -> pd.DataFrame:
"""
加载工单数据
Parameters
----------
product_col : str
产品类型列名
plan_col : str
计划加工时长列名
actual_col : str
实际加工时长列名
status_col : str
工单状态列名
Returns
-------
pd.DataFrame
"""
if not self.filepath.exists():
raise FileNotFoundError(f"文件不存在: {self.filepath}")
suffix = self.filepath.suffix.lower()
if suffix == ".csv":
self._raw_df = pd.read_csv(
self.filepath, encoding=self.encoding
)
elif suffix in (".xlsx", ".xls"):
self._raw_df = pd.read_excel(self.filepath)
else:
raise ValueError(f"不支持的文件格式: {suffix}")
# 列名标准化
col_aliases = {
product_col: ["产品类型", "product", "item_type", "型号"],
plan_col: ["计划时长", "plan_duration", "planned_time", "标准工时"],
actual_col: ["实际时长", "actual_duration", "real_time", "实际工时"],
status_col: ["状态", "status", "完成状态", "order_status"],
}
rename_map = {}
for target, aliases in col_aliases.items():
if target not in self._raw_df.columns:
for alias in aliases:
if alias in self._raw_df.columns:
rename_map[alias] = target
break
if rename_map:
self._raw_df = self._raw_df.rename(columns=rename_map)
return self._raw_df.copy()
def get_raw_data(self) -> Optional[pd.DataFrame]:
return self._raw_df.copy() if self._raw_df is not None else None
</details>
<details>
<summary></summary>
"""工单数据清洗与异常检测"""
import numpy as np
import pandas as pd
from typing import Tuple, Optional
class DataCleaner:
"""
FMS 工单数据清洗器
处理空值、过滤未完成工单、检测异常加工时长。
"""
def __init__(self, iqr_multiplier: float = 1.5):
self.iqr_multiplier = iqr_multiplier
def clean(self,
df: pd.DataFrame,
product_col: str = "product_type",
plan_col: str = "plan_time",
actual_col: str = "actual_time",
status_col: str = "status") -> pd.DataFrame:
"""
清洗工单数据
Parameters
----------
df : pd.DataFrame
原始工单数据
product_col : str
产品类型列名
plan_col : str
计划时长列名
actual_col : str
实际时长列名
status_col : str
状态列名
Returns
-------
pd.DataFrame
清洗后的数据
"""
result = df.copy()
# 确保数值列是 float
result[plan_col] = pd.to_numeric(result[plan_col], errors="coerce")
result[actual_col] = pd.to_numeric(result[actual_col], errors="coerce")
# 过滤未完成工单
if status_col in result.columns:
completed_mask = result[status_col].astype(str).str.contains(
"完成|done|complete|closed", case=False, na=False
)
result = result[completed_mask]
# 删除关键字段为空的行
result = result.dropna(subset=[product_col, actual_col])
# 过滤非正数
result = result[
(result[actual_col] > 0) &
(result[plan_col] > 0)
]
return result.reset_index(drop=True)
def detect_outliers(self,
df: pd.DataFrame,
group_col: str = "product_type",
value_col: str = "actual_time") -> pd.DataFrame:
"""
使用 IQR 法则检测异常加工时长
Parameters
----------
df : pd.DataFrame
清洗后的数据
group_col : str
分组列名
value_col : str
数值列名
Returns
-------
pd.DataFrame
新增 is_outlier 列
"""
result = df.copy()
result["is_outlier"] = False
for group_name, group_df in result.groupby(group_col):
values = group_df[value_col].values
q1, q3 = np.percentile(values, [25, 75])
iqr = q3 - q1
lower = q1 - self.iqr_multiplier * iqr
upper = q3 + self.iqr_multiplier * iqr
mask = (group_df.index.isin(group_df.index)) & (
(group_df[value_col] < lower) |
(group_df[value_col] > upper)
)
result.loc[group_df.index, "is_outlier"] = mask
return result
</details>
<details>
<summary></summary>
"""加工时长分析器"""
import numpy as np
import pandas as pd
from typing import Dict, Optional
class DurationAnalyzer:
"""
FMS 工单加工时长分析
按产品类型分组计算统计指标。
"""
def __init__(self,
product_col: str = "product_type",
plan_col: str = "plan_time",
actual_col: str = "actual_time"):
self.product_col = product_col
self.plan_col = plan_col
self.actual_col = actual_col
def analyze(self, df: pd.DataFrame) -> pd.DataFrame:
"""
按产品类型分析加工时长
Parameters
----------
df : pd.DataFrame
清洗后的数据
Returns
-------
pd.DataFrame
每种产品的统计指标
"""
summary = df.groupby(self.product_col).agg(
count=(self.actual_col, "count"),
avg_actual=(self.actual_col, "mean"),
std_actual=(self.actual_col, "std"),
min_actual=(self.actual_col, "min"),
max_actual=(self.actual_col, "max"),
avg_plan=(self.plan_col, "mean"),
).reset_index()
# 计算偏差
summary["deviation"] = (
summary["avg_actual"] - summary["avg_plan"]
)
summary["deviation_pct"] = (
summary["deviation"] / summary["avg_plan"] * 100
)
# 填充空值
summary["std_actual"] = summary["std_actual"].fillna(0)
# 排序
summary = summary.sort_values("avg_actual", ascending=False)
return summary
def analyze_excluding_outliers(self,
df: pd.DataFrame,
outlier_col: str = "is_outlier") -> pd.DataFrame:
"""
排除异常值后重新分析
Parameters
----------
df : pd.DataFrame
含异常标记的数据
outlier_col : str
异常标记列名
Returns
-------
pd.DataFrame
排除异常后的统计结果
"""
clean_df = df[~df[outlier_col]].copy()
return self.analyze(clean_df)
def get_recommendation(self, summary: pd.DataFrame) -> pd.DataFrame:
"""
生成排产建议（使用平均时长 + 1 个标准差作为安全时长）
Parameters
----------
summary : pd.DataFrame
分析结果
Returns
-------
pd.DataFrame
含排产建议时长的数据
"""
rec = summary.copy()
rec["suggested_time"] = rec["avg_actual"] + rec["std_actual"]
return rec[[self.product_col, "avg_actual", "std_actual",
"suggested_time", "count"]]
</details>
<details>
<summary></summary>
"""可视化器"""
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from pathlib import Path
from typing import Optional
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
class Visualizer:
"""FMS 工单加工时长可视化"""
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def plot_avg_duration_bar(self,
summary: pd.DataFrame,
product_col: str = "product_type",
value_col: str = "avg_actual",
title: str = "各类产品平均加工时长") -> None:
"""绘制平均加工时长柱状图"""
fig, ax = plt.subplots(figsize=(10, 5))
products = summary[product_col].tolist()
values = summary[value_col].tolist()
colors = plt.cm.Set3(np.linspace(0, 1, len(products)))
bars = ax.bar(products, values, color=colors)
ax.set_ylabel("平均加工时长 (分钟)", fontsize=12)
ax.set_title(title, fontsize=14, fontweight="bold")
ax.tick_params(axis="x", rotation=30)
# 标注数值
for bar, val in zip(bars, values):
ax.text(
bar.get_x() + bar.get_width() / 2,
bar.get_height() + 1,
f"{val:.1f}",
ha="center",
va="bottom",
fontsize=9,
)
plt.tight_layout()
plt.savefig(self.results_dir / "avg_duration_bar.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_duration_boxplot(self,
df: pd.DataFrame,
product_col: str = "product_type",
value_col: str = "actual_time") -> None:
"""绘制加工时长箱线图"""
fig, ax = plt.subplots(figsize=(10, 5))
products = df[product_col].unique()
data = [df[df[product_col] == p][value_col].values
for p in products]
bp = ax.boxplot(data, labels=products, patch_artist=True,
showfliers=True)
ax.set_ylabel("加工时长 (分钟)", fontsize=12)
ax.set_title("各类产品加工时长分布", fontsize=14, fontweight="bold")
ax.tick_params(axis="x", rotation=30)
# 着色
colors = plt.cm.Set3(np.linspace(0, 1, len(products)))
for patch, color in zip(bp["boxes"], colors):
patch.set_facecolor(color)
plt.tight_layout()
plt.savefig(self.results_dir / "duration_boxplot.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_plan_vs_actual(self,
summary: pd.DataFrame,
product_col: str = "product_type") -> None:
"""绘制计划 vs 实际对比图"""
fig, ax = plt.subplots(figsize=(10, 5))
products = summary[product_col].tolist()
x = np.arange(len(products))
width = 0.35
ax.bar(x - width/2, summary["avg_plan"], width,
label="计划时长", color="#3498DB", alpha=0.8)
ax.bar(x + width/2, summary["avg_actual"], width,
label="实际时长", color="#E74C3C", alpha=0.8)
ax.set_ylabel("时长 (分钟)", fontsize=12)
ax.set_title("计划 vs 实际加工时长对比", fontsize=14, fontweight="bold")
ax.set_xticks(x)
ax.set_xticklabels(products, rotation=30)
ax.legend()
plt.tight_layout()
plt.savefig(self.results_dir / "plan_vs_actual.png",
dpi=150, bbox_inches="tight")
plt.close()
</details>
<details>
<summary></summary>
"""合成数据生成器"""
import numpy as np
import pandas as pd
from pathlib import Path
from typing import Optional
from datetime import datetime, timedelta
class SyntheticDataGenerator:
"""
FMS 工单合成数据生成器
生成包含不同产品类型、计划/实际时长、异常值的模拟工单数据。
"""
def __init__(self, rng: Optional[np.random.RandomState] = None):
self.rng = rng or np.random.RandomState(42)
def generate(self,
n_records: int = 400,
start_date: str = "2025-01-01",
output_path: str = "fms_workorders.csv") -> pd.DataFrame:
"""
生成合成工单数据
Parameters
----------
n_records : int
工单条数
start_date : str
起始日期
output_path : str
输出路径
Returns
-------
pd.DataFrame
"""
# 产品类型配置：(名称, 计划时长均值, 实际时长均值, 标准差)
products = [
("A型法兰", 90, 110, 18),
("B型壳体", 70, 85, 12),
("C型轴套", 45, 55, 8),
("D型端盖", 60, 72, 10),
("E型支架", 100, 120, 22),
]
start = datetime.strptime(start_date, "%Y-%m-%d")
records = []
for _ in range(n_records):
prod_name, plan_mean, actual_mean, std = products[
self.rng.randint(0, len(products))
]
# 计划时长（略有波动）
plan_time = max(10, plan_mean + self.rng.normal(0, 5))
# 实际时长（正态分布 + 少量异常值）
if self.rng.random() < 0.05: # 5% 异常
actual_time = actual_mean + self.rng.uniform(2, 4) * std
else:
actual_time = max(5, self.rng.normal(actual_mean, std))
# 状态（95% 完成）
status = "完成" if self.rng.random() < 0.95 else "进行中"
# 日期
date = start + timedelta(days=int(self.rng.randint(0, 60)))
records.append({
"workorder_id": f"WO{_+1:05d}",
"product_type": prod_name,
"plan_time": round(plan_time, 1),
"actual_time": round(actual_time, 1),
"status": status,
"operator": f"OP{self.rng.randint(1, 8):02d}",
"date": date.strftime("%Y-%m-%d"),
})
df = pd.DataFrame(records)
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
df.to_csv(output_path, index=False, encoding="utf-8")
return df
</details>
<details>
<summary></summary>
"""
FMS 车间工单加工时长分析
================================================================================
课程映射（滨州职业学院《先进制造技术》）：
柔性制造系统FMS与先进生产管理：生产计划与排程（工单加工时长统计）
先进制造技术基础：制造系统性能评估（加工效率分析）
智能制造与数字孪生：制造大数据驱动决策（数据驱动的排产优化）
数控加工与CAD/CAM技术：数控加工工艺规划（工艺参数与加工时间关联）
技术栈（严格）：
pandas # 工单加载、分组统计
numpy # 统计计算与合成数据生成
matplotlib # 柱状图、箱线图、对比图可视化
"""
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import numpy as np
import pandas as pd
from pathlib import Path
from fms_workorder_analysis.workorder_loader import WorkorderLoader
from fms_workorder_analysis.data_cleaner import DataCleaner
from fms_workorder_analysis.duration_analyzer import DurationAnalyzer
from fms_workorder_analysis.visualizer import Visualizer
from fms_workorder_analysis.synthetic_data import SyntheticDataGenerator
def main():
"""主程序"""
print("=" * 70)
print("FMS 车间工单加工时长分析")
print("=" * 70)
results_dir = Path("results")
results_dir.mkdir(exist_ok=True)
# 1. 生成合成数据
print("\n[1/6] 生成合成工单数据...")
generator = SyntheticDataGenerator(rng=np.random.RandomState(42))
csv_path = "fms_workorders.csv"
df_raw = generator.generate(
n_records=400,
start_date="2025-01-01",
output_path=csv_path,
)
print(f" 生成 {len(df_raw)} 条工单记录")
# 2. 加载数据
print("\n[2/6] 加载工单数据...")
loader = WorkorderLoader(csv_path)
df = loader.load()
print(f" 加载 {len(df)} 条记录")
print(f" 产品类型: {df['product_type'].unique().tolist()}")
# 3. 数据清洗
print("\n[3/6] 清洗数据...")
cleaner = DataCleaner(iqr_multiplier=1.5)
df_clean = cleaner.clean(df)
print(f" 清洗后: {len(df_clean)} 条有效记录")
print(f" 各产品数量:")
for prod, count in df_clean["product_type"].value_counts().items():
print(f" {prod}: {count}")
# 4. 异常检测
print("\n[4/6] 检测异常加工时长...")
df_with_outliers = cleaner.detect_outliers(
df_clean, group_col="product_type", valu
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！