python的先进制造技术工业场景模拟第十篇:读取机器人工作站生产记录,统计每日工件抓取成功数与抓取失败数量。
周三夜班,机器人焊接工作站。
"今晚又废了三个件,"夜班班长老赵指着废料筐,"机器人抓件的时候,吸盘没吸住——工件在夹具上歪了,焊枪一过去,直接焊偏。三个壳体,每个两百多,六百多块钱没了。"
我调出机器人工作站的 HMI 日志。屏幕上滚动着一行行记录:时间戳、机器人编号、工件编号、抓取结果(SUCCESS / FAIL)、失败原因代码(01=吸盘真空不足、02=工件位置偏移、03=夹具干涉、04=视觉定位超时)、循环时间(秒)。
"系统里不是有抓取记录吗?"我问。
"有,"老赵点开一页,"但它是按'次'记录的——每次抓取一行,一晚上几千行。我想知道的是:今天总共抓了多少次、成功了几次、失败了几次、失败率多少、主要是哪种原因导致的。系统没有'每日抓取统计'这个功能——它只会实时显示当前这一次抓没抓上。"
"所以你现在每次开班,都不知道昨晚的抓取情况?"
"对,"老赵苦笑,"第二天早上我来接班,看到废料筐里多了几个废件,才知道昨晚抓取有问题。但那时候夜班已经结束了,没法实时干预。厂长说必须搞一个'每日抓取统计报表',每天早上贴在看板上——成功多少、失败多少、失败原因排名,一目了然。"
"明白了,"我打开 VS Code,"你需要一个程序:读取机器人工作站的生产记录 CSV,按日期分组,统计每天的抓取成功数和失败数,计算失败率,再按失败原因分类汇总。输出一张每日统计表和一张趋势图。这样每天早上,你直接把报表打印出来贴在看板上:昨天抓了 480 次,成功 465 次,失败 15 次,失败率 3.1%,主要原因是吸盘真空不足(9 次)。"
我敲了几行代码:
import pandas as pd
import matplotlib.pyplot as plt
# 读取抓取记录
df = pd.read_csv("robot_grasp_log.csv", parse_dates=["timestamp"])
# 提取日期
df["date"] = df["timestamp"].dt.date
# 按日期分组统计
daily = df.groupby("date").agg(
total=("grasp_result", "count"),
success=("grasp_result", lambda x: (x == "SUCCESS").sum()),
fail=("grasp_result", lambda x: (x == "FAIL").sum()),
).reset_index()
daily["fail_rate"] = daily["fail"] / daily["total"] * 100
# 按失败原因统计
fail_reasons = df[df["grasp_result"] == "FAIL"]["fail_reason"].value_counts()
print(daily)
print(fail_reasons)
"这只是核心逻辑,"我说,"完整版会用 OOP 封装:一个类管数据加载(解析时间戳、统一结果编码),一个类管数据清洗(处理缺失值、异常时间戳),一个类管每日统计(成功/失败计数、失败率计算),一个类管失败原因分类汇总,一个类管可视化(每日成功/失败堆叠柱状图、失败率趋势图、失败原因饼图)。数据自包含——用 numpy 合成一份机器人抓取日志 CSV,读者下载就能跑。"
老赵凑近屏幕:"所以你这东西,就是把几千行'每次抓了没抓上'的记录,按天汇总——成功多少、失败多少、失败原因是什么。一张表、一张图,早上看一眼就知道昨晚的抓取情况。"
"对。而且还能做趋势分析,"我补充,"比如连续三天失败率从 1% 升到 3% 又升到 5%——这说明吸盘系统在退化,不是偶发问题。可以提前安排维护,不等废件出来再处理。机器人工作站的管理不是'抓上了就行',是'用数据看趋势,在废件出现之前就发现问题'。"
一、实际应用场景(真实痛点)
场景设定:机器人工作站(焊接/搬运/装配)每次执行工件抓取动作后,在日志中记录时间戳、机器人编号、工件编号、抓取结果(成功/失败)和失败原因代码。一个班次产生数千条抓取记录。现场管理者需要每日统计抓取成功数、失败数、失败率及失败原因分布,用于评估工作站稳定性和安排设备维护。常见痛点:日志按"次"逐条记录,行数庞大,系统不提供按日聚合的"每日抓取统计"报表,依赖次日人工查看废料筐来反推抓取质量。
现场原话(叙事化):
"我不是不想管抓取质量,"老赵说,"每次抓没抓上,系统里都有记录。但它是流水账——一行一行滚过去,谁看得过来?我需要知道的是:今天总共抓了多少次、成功了几次、失败了几次、失败率多少、主要是哪种原因。这些系统给不了我。第二天早上我看到废料筐才知道昨晚有问题——但那时候已经废了三个件了。"
核心矛盾:"逐次记录的抓取流水账"与"管理者需要的每日聚合统计"之间的汇总断层。需要一个"机器人工作站抓取记录统计程序",用
"pandas" 按日期分组聚合,用
"numpy" 计算失败率和统计指标,用
"matplotlib" 绘制每日趋势图和失败原因分布图,输出结构化的每日抓取统计报表。
二、痛点分析(映射到滨州职业学院《先进制造技术》课程模型)
《先进制造技术》模块 本篇痛点对应
工业机器人技术基础:机器人末端执行器与抓取策略 抓取可靠性评估:基于日志数据的抓取成功率统计与失败模式分析。
FMS与先进生产管理:FMS 的监控与数据采集(MDC) 生产数据采集:从机器人日志中提取结构化指标,支撑生产管理决策。
智能制造与数字孪生:生产过程的实时监控 质量趋势可视化:用趋势图反映抓取质量的时序变化,实现早期预警。
先进制造新模式:数据驱动的持续改进 根因分析:按失败原因分类汇总,定位系统性问题而非偶发故障。
一句话总结:我们需要构建一个"机器人工作站抓取记录统计程序",用
"pandas" 按日期分组聚合,用
"numpy" 计算失败率,用
"matplotlib" 绘制趋势图和分布图,实现从逐次流水账到每日质量报表的转化。
三、核心逻辑讲解(大白话)
3.1 问题本质:把抓取记录想象成"快递配送记录"
把机器人抓取日志想象成"快递配送的签收记录":
* 每次抓取 = 一次配送:快递员每次送一个包裹,记录"送到了(成功)"或"没送到(失败)"。
* 失败原因 = 拒收原因:"收件人不在"(吸盘真空不足)、"地址不对"(工件位置偏移)、"电话不通"(视觉定位超时)。
* 每日统计 = 快递站长的日报表:今天总共派了 480 单,成功 465 单,失败 15 单,失败率 3.1%——其中 9 单是"收件人不在"(吸盘问题),4 单是"地址不对"(定位偏移),2 单是"电话不通"(视觉超时)。
* 趋势图 = 配送质量走势:连续三天失败率从 1% → 3% → 5%,说明快递员可能车子出问题了——对应车间里"吸盘系统退化,需要维护"。
工业应用:
* pandas 日期分组:
"df["date"] = df["timestamp"].dt.date" 提取日期,
"groupby("date").agg(...)" 按天聚合——一行代码完成几千条记录的每日汇总。
* numpy 失败率计算:
"fail_rate = fails / total * 100",向量化计算所有天的失败率,比逐天循环快几十倍。
* matplotlib 堆叠柱状图:横轴日期,柱子高度是总抓取次数,绿色部分是成功数,红色部分是失败数——一张图看清每天"抓了多少、废了多少"。
3.2 业务逻辑 → 代码映射
定义机器人抓取数据模型
│
▼ GraspDataLoader (pandas)
读取抓取日志 CSV:
pd.read_csv("robot_grasp_log.csv", parse_dates=["timestamp"])
解析时间戳、机器人编号、工件编号、抓取结果、失败原因
│
▼ DataCleaner (pandas)
数据清洗:
处理缺失时间戳(删除或标记)
统一结果编码(SUCCESS/FAIL → 1/0)
解析失败原因代码(01→吸盘真空不足, 02→工件偏移, ...)
│
▼ DailyStatistics (pandas/numpy)
每日统计:
groupby("date").agg(
total=count,
success=sum(result==1),
fail=sum(result==0)
)
fail_rate = fail / total * 100
│
▼ ReasonAnalyzer (pandas)
失败原因分类汇总:
fail_df.groupby("fail_reason").size()
计算各原因占比
│
▼ GraspVisualizer (matplotlib)
可视化:
1. 每日成功/失败堆叠柱状图
2. 每日失败率趋势折线图
3. 失败原因饼图
│
▼ SyntheticDataGenerator (numpy)
合成数据生成:
用 numpy 按泊松分布生成每日抓取次数
按概率分布模拟成功/失败和失败原因
确保数据自包含、可复现
3.3 为什么用 groupby 聚合而不是逐行循环?
* 问题:30 天 × 每天 500 条 = 15000 行记录,逐行判断日期再累加——代码长、速度慢。
* 处理策略:
"pandas.groupby()" +
"agg()" 向量化聚合,C 语言底层加速,毫秒级完成。
* 工程合理性:机器人工作站 7×24 小时运行,日志量可能达到百万级,向量化是唯一合理的选择。
3.4 分析前后对比
维度 人工查看废料筐 Python 程序统计
判断依据 废料筐里多了几个废件 日志中的 SUCCESS/FAIL 标记
时间粒度 次日才发现 实时/按班次统计
信息量 "废了几个" 成功数、失败数、失败率、原因分布
趋势感知 无 失败率连续上升 → 提前预警
根因定位 靠猜 失败原因排名 → 定位系统性问题
四、OOP 代码实现
4.1 项目结构
robot_grasp_analyzer/
├── robot_grasp_analyzer/
│ ├── __init__.py
│ ├── data_loader.py # 抓取日志数据加载
│ ├── data_cleaner.py # 数据清洗
│ ├── daily_statistics.py # 每日统计
│ ├── reason_analyzer.py # 失败原因分类
│ ├── visualizer.py # 可视化
│ └── synthetic_data.py # 合成数据生成
├── tests/
│ ├── __init__.py
│ └── test_grasp_analyzer.py # 单元测试
├── results/
│ ├── daily_report.csv # 每日统计报表
│ ├── reason_summary.csv # 失败原因汇总
│ ├── daily_stack.png # 每日堆叠柱状图
│ ├── fail_rate_trend.png # 失败率趋势图
│ ├── reason_pie.png # 失败原因饼图
│ └── grasp_report.txt # 文本分析报告
└── run_grasp.py # 主程序入口
4.2 核心源码
<details>
<summary></summary>
"""机器人抓取日志数据加载器"""
import pandas as pd
from pathlib import Path
from typing import Optional
class GraspDataLoader:
"""
机器人抓取日志数据加载器
支持 CSV 格式的抓取记录。
"""
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,
timestamp_col: str = "timestamp",
robot_col: str = "robot_id",
workpiece_col: str = "workpiece_id",
result_col: str = "grasp_result",
reason_col: str = "fail_reason") -> pd.DataFrame:
"""
加载抓取日志数据
Parameters
----------
timestamp_col : str
时间戳列名
robot_col : str
机器人编号列名
workpiece_col : str
工件编号列名
result_col : str
抓取结果列名
reason_col : str
失败原因列名
Returns
-------
pd.DataFrame
"""
if not self.filepath.exists():
raise FileNotFoundError(f"文件不存在: {self.filepath}")
self._raw_df = pd.read_csv(
self.filepath,
encoding=self.encoding,
parse_dates=[timestamp_col] if timestamp_col else None,
)
# 列名标准化
col_aliases = {
timestamp_col: ["时间", "timestamp", "time", "datetime"],
robot_col: ["机器人", "robot_id", "robot", "station"],
workpiece_col: ["工件", "workpiece_id", "part_id", "wp"],
result_col: ["结果", "grasp_result", "result", "status"],
reason_col: ["失败原因", "fail_reason", "reason", "error_code"],
}
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 Dict, Optional
class DataCleaner:
"""
机器人抓取数据清洗器
处理缺失值、统一结果编码、解析失败原因。
"""
# 结果编码映射
RESULT_MAP = {
"SUCCESS": 1, "成功": 1, "OK": 1, "Pass": 1, "1": 1,
"FAIL": 0, "失败": 0, "NG": 0, "Fail": 0, "0": 0,
}
# 失败原因代码映射
REASON_MAP = {
"01": "吸盘真空不足",
"02": "工件位置偏移",
"03": "夹具干涉",
"04": "视觉定位超时",
"05": "气压不足",
"06": "工件变形",
"Unknown": "未知原因",
"": "未知原因",
}
def __init__(self):
pass
def clean(self,
df: pd.DataFrame,
timestamp_col: str = "timestamp",
result_col: str = "grasp_result",
reason_col: str = "fail_reason") -> pd.DataFrame:
"""
清洗抓取数据
Parameters
----------
df : pd.DataFrame
原始数据
timestamp_col : str
时间戳列名
result_col : str
抓取结果列名
reason_col : str
失败原因列名
Returns
-------
pd.DataFrame
"""
result = df.copy()
# 时间戳处理
if timestamp_col in result.columns:
result[timestamp_col] = pd.to_datetime(
result[timestamp_col], errors="coerce"
)
# 删除无效时间戳
result = result.dropna(subset=[timestamp_col])
# 提取日期
result["date"] = result[timestamp_col].dt.date.astype(str)
# 统一结果编码
if result_col in result.columns:
result["result_code"] = result[result_col].map(
lambda x: self.RESULT_MAP.get(str(x).strip(), -1)
)
else:
result["result_code"] = -1
# 解析失败原因
if reason_col in result.columns:
result["reason_text"] = result[reason_col].map(
lambda x: self.REASON_MAP.get(str(x).strip(), str(x))
)
else:
result["reason_text"] = "未知原因"
return result.reset_index(drop=True)
</details>
<details>
<summary></summary>
"""每日抓取统计"""
import numpy as np
import pandas as pd
from typing import Dict, Optional
class DailyStatistics:
"""
机器人每日抓取统计器
按日期分组,统计成功数、失败数、失败率。
"""
def __init__(self):
pass
def calculate(self,
df: pd.DataFrame,
date_col: str = "date",
result_col: str = "result_code") -> pd.DataFrame:
"""
计算每日统计
Parameters
----------
df : pd.DataFrame
清洗后的数据
date_col : str
日期列名
result_col : str
结果编码列名(1=成功, 0=失败)
Returns
-------
pd.DataFrame
每日统计表
"""
if df.empty:
return pd.DataFrame()
# 按日期分组聚合
daily = df.groupby(date_col).agg(
total=(result_col, "count"),
success=(result_col, lambda x: (x == 1).sum()),
fail=(result_col, lambda x: (x == 0).sum()),
).reset_index()
# 计算失败率
daily["fail_rate"] = (daily["fail"] / daily["total"] * 100).round(2)
daily["success_rate"] = (100 - daily["fail_rate"]).round(2)
# 排序
daily = daily.sort_values(date_col).reset_index(drop=True)
return daily
def get_overall_summary(self, daily_df: pd.DataFrame) -> Dict:
"""计算整体汇总"""
if daily_df.empty:
return {}
return {
"total_days": len(daily_df),
"total_grasps": daily_df["total"].sum(),
"total_success": daily_df["success"].sum(),
"total_fail": daily_df["fail"].sum(),
"overall_fail_rate": (
daily_df["fail"].sum() / daily_df["total"].sum() * 100
).round(2),
"avg_daily_grasps": daily_df["total"].mean().round(1),
"max_daily_fail_rate": daily_df["fail_rate"].max(),
"min_daily_fail_rate": daily_df["fail_rate"].min(),
}
</details>
<details>
<summary></summary>
"""失败原因分类分析"""
import numpy as np
import pandas as pd
from typing import Dict, Optional
class ReasonAnalyzer:
"""
机器人抓取失败原因分类分析器
按失败原因分组汇总,计算各原因占比。
"""
def __init__(self):
pass
def analyze(self,
df: pd.DataFrame,
reason_col: str = "reason_text") -> pd.DataFrame:
"""
分析失败原因分布
Parameters
----------
df : pd.DataFrame
清洗后的数据(含失败原因文本)
reason_col : str
失败原因列名
Returns
-------
pd.DataFrame
按数量降序排列的原因汇总表
"""
# 只取失败记录
fail_df = df[df["result_code"] == 0].copy()
if fail_df.empty:
return pd.DataFrame()
# 按原因分组
reason_counts = fail_df[reason_col].value_counts().reset_index()
reason_counts.columns = ["fail_reason", "count"]
# 计算占比
total_fails = reason_counts["count"].sum()
reason_counts["percentage"] = (
reason_counts["count"] / total_fails * 100
).round(2)
# 排序
reason_counts = reason_counts.sort_values(
"count", ascending=False
).reset_index(drop=True)
return reason_counts
def get_top_reason(self, reason_df: pd.DataFrame) -> Optional[str]:
"""获取最主要的失败原因"""
if reason_df.empty:
return None
return reason_df.iloc[0]["fail_reason"]
</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 GraspVisualizer:
"""机器人抓取统计可视化"""
COLORS = {
"success": "#2ECC71",
"fail": "#E74C3C",
"warning": "#F39C12",
}
PIE_COLORS = [
"#E74C3C", "#F39C12", "#3498DB", "#9B59B6",
"#1ABC9C", "#34495E", "#E67E22", "#95A5A6",
]
def __init__(self, results_dir: str = "results"):
self.results_dir = Path(results_dir)
self.results_dir.mkdir(exist_ok=True)
def plot_daily_stack(self,
df: pd.DataFrame,
date_col: str = "date",
success_col: str = "success",
fail_col: str = "fail",
title: str = "每日抓取成功/失败统计") -> None:
"""绘制每日成功/失败堆叠柱状图"""
fig, ax = plt.subplots(figsize=(14, 6))
x = np.arange(len(df))
width = 0.6
bars_success = ax.bar(
x, df[success_col], width,
label="成功", color=self.COLORS["success"], alpha=0.85,
)
bars_fail = ax.bar(
x, df[fail_col], width,
bottom=df[success_col],
label="失败", color=self.COLORS["fail"], alpha=0.85,
)
# 标注失败数
for i, (s, f) in enumerate(zip(df[success_col], df[fail_col])):
if f > 0:
ax.text(
i, s + f / 2,
str(f), ha="center", va="center",
fontsize=8, fontweight="bold", color="white",
)
ax.set_xlabel("日期", fontsize=12)
ax.set_ylabel("抓取次数", fontsize=12)
ax.set_title(title, fontsize=14, fontweight="bold")
ax.set_xticks(x)
ax.set_xticklabels(df[date_col], rotation=45, ha="right")
ax.legend()
ax.grid(axis="y", alpha=0.3)
plt.tight_layout()
plt.savefig(self.results_dir / "daily_stack.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_fail_rate_trend(self,
df: pd.DataFrame,
date_col: str = "date",
fail_rate_col: str = "fail_rate",
title: str = "每日失败率趋势") -> None:
"""绘制每日失败率趋势折线图"""
fig, ax = plt.subplots(figsize=(14, 6))
ax.plot(
df[date_col], df[fail_rate_col],
marker="o", linewidth=2, color="#E74C3C", markersize=6,
)
# 参考线
ax.axhline(y=5.0, color="#F39C12", linestyle="--",
linewidth=1.5, alpha=0.7, label="警告线 5%")
ax.axhline(y=10.0, color="#E74C3C", linestyle="--",
linewidth=1.5, alpha=0.7, label="危险线 10%")
# 标注数值
for _, row in df.iterrows():
ax.annotate(
f"{row[fail_rate_col]:.1f}%",
(row[date_col], row[fail_rate_col]),
textcoords="offset points", xytext=(0, 10),
ha="center", fontsize=8,
)
ax.set_xlabel("日期", fontsize=12)
ax.set_ylabel("失败率 (%)", fontsize=12)
ax.set_title(title, fontsize=14, fontweight="bold")
ax.legend()
ax.grid(alpha=0.3)
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
plt.savefig(self.results_dir / "fail_rate_trend.png",
dpi=150, bbox_inches="tight")
plt.close()
def plot_reason_pie(self,
df: pd.DataFrame,
reason_col: str = "fail_reason",
count_col: str = "count",
title: str = "失败原因分布") -> None:
"""绘制失败原因饼图"""
if df.empty:
return
fig, ax = plt.subplots(figsize=(10, 8))
wedges, texts, autotexts = ax.pie(
df[count_col],
labels=df[reason_col],
autopct="%1.1f%%",
colors=self.PIE_COLORS[:len(df)],
startangle=90,
pctdistance=0.85,
)
# 美化
for autotext in autotexts:
autotext.set_color("white")
autotext.set_fontsize(10)
autotext.set_fontweight("bold")
ax.set_title(title, fontsize=14, fontweight="bold")
plt.tight_layout()
plt.savefig(self.results_dir / "reason_pie.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
class SyntheticDataGenerator:
"""
机器人抓取日志合成数据生成器
模拟多天、多机器人的抓取记录。
"""
def __init__(self, rng: Optional[np.random.RandomState] = None):
self.rng = rng or np.random.RandomState(42)
self.robots = [f"Robot-{i}" for i in range(1, 4)] # 3 台机器人
# 失败原因及概率分布
self.fail_reasons = {
"01": 0.40, # 吸盘真空不足(最常见)
"02": 0.25, # 工件位置偏移
"03": 0.15, # 夹具干涉
"04": 0.10, # 视觉定位超时
"05": 0.07, # 气压不足
"06": 0.03, # 工件变形
}
def generate(self,
n_days: int = 30,
output_path: str = "robot_grasp_log.csv") -> pd.DataFrame:
"""
生成合成抓取日志
Parameters
----------
n_days : int
天数
output_path : str
输出路径
Returns
-------
pd.DataFrame
"""
records = []
record_id = 0
base_date = pd.Timestamp("2026-01-01")
for day in range(n_days):
current_date = base_date + pd.Timedelta(days=day)
for robot in self.robots:
# 每天每台机器人约 400-600 次抓取(泊松分布)
daily_grasps = self.rng.poisson(lam=500)
# 模拟失败率趋势(随时间略有上升,模拟设备退化)
base_fail_rate = 0.02 + day * 0.001 # 从 2% 缓慢上升到 ~5%
for _ in range(daily_grasps):
record_id += 1
# 时间戳(均匀分布在 8 小时班次内)
second_in_shift = self.rng.randint(0, 8 * 3600)
timestamp = current_date + pd.Timedelta(seconds=second_in_shift)
# 抓取结果
is_fail = self.rng.random() < base_fail_rate
grasp_result = "FAIL" if is_fail else "SUCCESS"
# 失败原因
fail_reason = ""
if is_fail:
# 按概率分布选择失败原因
rand = self.rng.random()
cumsum = 0
for reason, prob in self.fail_reasons.items():
cumsum += prob
if rand <= cumsum:
fail_reason = reason
break
if not fail_reason:
fail_reason = "Unknown"
# 工件编号
利用AI解决实际问题,如果你觉得这个工具好用,欢迎关注长安牧笛!
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐

所有评论(0)