企业官网建设流程全解析

周三下午，FMS 柔性制造单元控制室。

"这条线，两台加工中心加一台机器人，跑了一周，"生产主管老周指着 MES 屏幕上的设备状态条，"MES 只记了每个时刻的状态——'加工'、'对刀'、'夹具调整'、'待机'，但没人算过每种状态到底占了多长时间。我想知道加工时间占比多少、对刀和夹具调整是不是太长了、待机是不是因为上下料节拍没匹配好。"

我点开导出的状态日志 CSV。

"这表里有什么？"老周问。

"每条记录一行，有时间戳、设备编号、状态类型，"我指着数据，"问题是状态是离散跳变的——8:01:23 加工，8:15:47 待机，8:16:02 对刀……时间不连续，要算每个状态的总时长，得相邻两条做差，还要处理跨天、状态缺失、设备编号不统一这些脏数据。"

"我就想干一件事，"老周说，"把每台设备、每种状态（加工/对刀/夹具调整/待机）的累计时长算出来，算占比，画个甘特图看时间线，再看看不同设备之间状态分布是不是均衡。如果 1 号机加工占 65%，2 号机只有 40%，那要么 1 号机活太多，要么 2 号机辅助时间太长。"

"比如 1 号机，总运行 8 小时，"我接话，"加工 3.2 小时（40%），对刀 0.8 小时（10%），夹具调整 1.2 小时（15%），待机 2.8 小时（35%）。待机占比 35% 明显偏高——查一下是机器人送料慢了还是托盘交换时间太长。对刀 10% 也偏高，正常应该 3~5%，可能是刀具补偿参数没固化，每次都要重新对。"

"对，"老周点头，"还有，我想看状态转换频率——加工完是转待机多还是直接转对刀多？如果频繁在加工和对刀之间跳，说明刀具寿命管理有问题，经常打到刀。最好还能用聚类把设备按状态分布分个组，看看是不是有设备'病态'运行。"

"用 pandas 按设备+状态分组做时间差聚合，numpy 算占比，matplotlib 画堆叠条形图+甘特时间线，scipy 做状态时长分布检验，scikit-learn 聚类设备状态画像，networkx 建设备-状态转移关系网，"我开工程，"数据自包含，合成一批 FMS 状态日志，下载就能跑。"

敲了行原型：

df["duration"] = df.groupby("device_id")["timestamp"].diff()

state_time = df.groupby(["device_id","state"])["duration"].sum().unstack()

"完整版用 OOP 封好，"我说，"一个类管数据加载和时间差计算，一个类做状态时长聚合，一个类画甘特图，一个类做统计检验，一个类聚类设备画像，一个类建状态转移网，一个类出图。输出设备状态统计表、状态转移矩阵、聚类结果，存 results/。"

老周凑近看："那以后看报告：2 台加工中心+1 台机器人，1 号机加工占比 62% 正常，2 号机只有 38%——一查，夹具调整占了 28%，因为第二工位的液压夹具响应慢了 8 秒，每循环多 16 秒辅助时间。状态转移分析发现 2 号机'加工→对刀'跳转频率是 1 号机的 3 倍，查了是刀库第二把刀的 Z 轴原点漂移。聚类把 2 号机单独分一组，状态画像明显异常。调整后 2 号机加工占比提到 58%。"

"对，"我接话，"设备状态不是看'在不在运行'，是看时间都花在哪了。辅助时间吃掉的是真金白银的产能。数字孪生里要建 OEE 模型，状态时长分布就是底层输入。"

一、实际应用场景（真实痛点）

场景设定：FMS 柔性制造单元（2~4 台加工中心 + 机器人上下料 + 托盘交换系统），MES 系统每秒/每状态变更记录设备状态日志。生产主管需要量化每台设备的有效加工时间占比，识别辅助时间过长、待机过多的瓶颈设备，优化单元级生产节拍。

现场原话（叙事化）：

"不是我们不想提高 OEE，"老周说，"是以前只看'设备开没开'。MES 上绿灯亮着就觉得在干活。后来我手动算了一次——把状态日志导出来，Excel 里一条条减时间——发现绿灯里有一半是'待机'，机器人还没把料送过来。那这绿灯有什么意义？我想让数据自动告诉我：每台设备的时间到底怎么花的，哪块是浪费。"

"还有夹具调整的问题，"老周补充，"第二工位换型时夹具调整要 15 分钟，第一工位只要 5 分钟。同样的产品，同样的夹具，为什么差 10 分钟？后来发现是液压阀响应慢，但 MES 里只记'夹具调整'，不区分是哪个工位、哪个夹具。这种'状态相同但耗时不同'的问题，不聚合分析根本发现不了。"

核心矛盾："离散状态流水日志" 与 "设备级状态时长聚合 + 占比分析 + 状态转移模式 + 设备画像聚类 + 瓶颈定位" 之间的断层。

二、痛点分析（映射到滨州职业学院《先进制造技术》课程模型）

《先进制造技术》模块 本篇痛点对应

柔性制造系统FMS与先进生产管理：设备综合效率(OEE)、单元调度 设备状态时长统计与辅助时间分析

先进制造技术基础：生产效能评估、时间研究 状态占比量化与瓶颈识别

智能制造与数字孪生：OEE 数字孪生模型数据底座 状态时长分布作为孪生输入

数控加工与CAD/CAM技术：刀具管理与对刀策略 对刀时长异常检测

先进制造新模式：精益生产、消除浪费 待机与辅助时间压缩

一句话总结：我们需要一个"FMS 加工单元设备状态时长统计与瓶颈分析程序"，用

"pandas" 做时间差聚合，

"numpy" 算占比，

"matplotlib" 画堆叠条形图+甘特图，

"scipy" 做分布检验，

"scikit-learn" 聚类设备状态画像，

"networkx" 建设备-状态转移关系网，实现从"状态日志流水"到"时长量化 + 占比分析 + 转移模式 + 设备画像"。

三、核心逻辑讲解（大白话）

3.1 问题本质：把设备想成"上班打卡"

把 FMS 设备想成一个员工的一天：

* 加工 = 实际干活（产出价值）

* 对刀 = 准备工作（磨刀不误砍柴工，但磨太久就是浪费）

* 夹具调整 = 换工位/换模具（必要的，但应该尽量短）

* 待机 = 等活干（机器人没送料、前道工序没完、没毛坯）

* 状态日志 = 打卡记录（8:00 开始干活，8:45 去对刀，8:50 回来接着干）

* 时长统计 = 算每件事花了多久

* 占比分析 = 一天里真正干活的时间占多少

* 状态转移 = 干完活是去对刀还是等活（频繁跳转说明不稳定）

* 设备画像聚类 = 哪些员工的工作模式像（把"摸鱼"的挑出来）

3.2 业务逻辑 → 代码映射

导入设备状态日志

│

▼ StateLoader (pandas)

读取 CSV：

timestamp, device_id, state,

(可选) station, fixture_id, tool_id

解析时间、排序、相邻时间差

│

▼ StateDurationCalculator (pandas/numpy)

时长计算：

每条记录 duration = 下一条 timestamp - 当前 timestamp

按 device_id × state 聚合 sum(duration)

计算占比 = 状态时长 / 总运行时长

│

▼ StateTransitionAnalyzer

状态转移矩阵：

统计 加工→待机、加工→对刀 等转移频次

计算转移概率

│

▼ StateStatistics (scipy)

统计检验：

设备间加工时长占比 t 检验/ANOVA

状态时长分布正态性检验

│

▼ DeviceProfiler (scikit-learn)

设备状态画像聚类：

特征=[加工占比, 对刀占比, 夹具调整占比, 待机占比, 转移频率]

KMeans 分群

│

▼ StateGraph (networkx)

设备-状态转移关系网：

节点=设备/状态

边=转移关系, 边权=转移频次

│

▼ StateVisualizer (matplotlib)

可视化：

1. 设备状态堆叠条形图(加工/对刀/夹具/待机)

2. 甘特时间线图(每台设备状态随时间变化)

3. 状态转移热力图

4. 设备画像聚类散点图

5. 设备-状态关系网

│

▼ SyntheticStateGenerator (numpy)

合成数据：

3设备×4状态×多天运行

含瓶颈设备(辅助时间过长)

3.3 为什么不能只看"设备开没开"

视角 问题

设备绿灯 = 在运行 绿灯里可能 35% 是待机

加工时间 = 总运行时间 忽略辅助时间吃掉产能

状态时长分解 精确到分钟，定位浪费在哪

转移模式 频繁跳转 = 工艺不稳定

3.4 分析前后对比

维度 MES 状态灯 本程序

加工占比 不知道 精确到百分比

辅助时间 "感觉长" 量化+对比

瓶颈设备 凭经验 聚类+统计检验

改进方向 模糊 数据指向具体原因

四、OOP 代码实现

4.1 项目结构

fms_state_analysis/

├── fms_state_analysis/

│ ├── __init__.py

│ ├── state_loader.py # 状态日志加载

│ ├── duration_calculator.py # 时长计算

│ ├── transition_analyzer.py # 状态转移分析

│ ├── state_statistics.py # 统计检验(scipy)

│ ├── device_profiler.py # 设备画像聚类(sklearn)

│ ├── state_graph.py # 设备-状态关系网(networkx)

│ ├── visualizer.py # 可视化

│ └── synthetic_data.py # 合成数据

├── tests/

│ ├── __init__.py

│ └── test_state_analysis.py

├── results/

│ ├── state_stacked_bar.png

│ ├── gantt_timeline.png

│ ├── transition_heatmap.png

│ ├── device_cluster.png

│ ├── state_network.png

│ ├── device_state_summary.csv

│ ├── transition_matrix.csv

│ ├── device_clusters.csv

│ └── state_report.txt

└── run_state_analysis.py

4.2 核心源码

<details>

<summary></summary>

"""FMS设备状态日志加载器"""

import pandas as pd

from pathlib import Path

from typing import Optional

class StateLoader:

"""加载设备状态日志CSV"""

VALID_STATES = ["加工", "对刀", "夹具调整", "待机"]

def __init__(self, filepath: str = "state_log.csv",

encoding: str = "utf-8"):

self.filepath = Path(filepath)

self.encoding = encoding

self._raw: Optional[pd.DataFrame] = None

def load(self) -> pd.DataFrame:

if not self.filepath.exists():

raise FileNotFoundError(f"文件不存在: {self.filepath}")

self._raw = pd.read_csv(self.filepath, encoding=self.encoding)

rename = {}

for tgt, al in {

"timestamp": ["timestamp", "时间", "time"],

"device_id": ["device_id", "设备编号", "device"],

"state": ["state", "状态", "status"],

"station": ["station", "工位", "station_id"],

"fixture_id": ["fixture_id", "夹具编号", "fixture"],

"tool_id": ["tool_id", "刀具编号", "tool"],

}.items():

if tgt not in self._raw.columns:

for a in al:

if a in self._raw.columns:

rename[a] = tgt

break

self._raw = self._raw.rename(columns=rename)

req = ["timestamp", "device_id", "state"]

miss = [c for c in req if c not in self._raw.columns]

if miss:

raise ValueError(f"缺少必要列: {miss}")

# 时间解析

self._raw["timestamp"] = pd.to_datetime(

self._raw["timestamp"], errors="coerce"

)

self._raw = self._raw.dropna(subset=["timestamp"]).sort_values(

["device_id", "timestamp"]

).reset_index(drop=True)

# 状态标准化

self._raw["state"] = self._raw["state"].astype(str).str.strip()

state_map = {}

for s in self._raw["state"].unique():

for vs in self.VALID_STATES:

if vs in s or s in vs:

state_map[s] = vs

break

if s not in state_map:

state_map[s] = "待机"

self._raw["state"] = self._raw["state"].map(state_map).fillna("待机")

self._raw["device_id"] = self._raw["device_id"].astype(str).str.strip()

for c in ["station", "fixture_id", "tool_id"]:

if c in self._raw.columns:

self._raw[c] = self._raw[c].fillna("N/A").astype(str).str.strip()

return self._raw.reset_index(drop=True)

</details>

<details>

<summary></summary>

"""状态时长计算器"""

import numpy as np

import pandas as pd

from typing import Optional

class DurationCalculator:

"""计算每台设备每种状态的累计时长"""

def __init__(self):

pass

def compute_duration(self, df: pd.DataFrame) -> pd.DataFrame:

"""计算每条记录到下一跳的时长(秒)"""

out = df.copy()

out["duration_sec"] = 0.0

for dev, grp in out.groupby("device_id"):

idx = grp.index

diffs = grp["timestamp"].diff().dt.total_seconds()

# 最后一条用前一条平均时长填充

avg_dur = diffs.dropna().median()

if pd.isna(avg_dur) or avg_dur <= 0:

avg_dur = 60.0

diffs = diffs.fillna(avg_dur)

out.loc[idx, "duration_sec"] = diffs.shift(-1).fillna(avg_dur)

out["duration_min"] = (out["duration_sec"] / 60.0).round(2)

return out

def aggregate(self, df: pd.DataFrame) -> pd.DataFrame:

"""按设备×状态聚合"""

if "duration_min" not in df.columns:

df = self.compute_duration(df)

agg = df.groupby(["device_id", "state"])["duration_min"].sum().reset_index()

pivot = agg.pivot(index="device_id", columns="state", values="duration_min").fillna(0)

# 确保四列齐全

for s in ["加工", "对刀", "夹具调整", "待机"]:

if s not in pivot.columns:

pivot[s] = 0.0

pivot["total_min"] = pivot.sum(axis=1)

for s in ["加工", "对刀", "夹具调整", "待机"]:

pivot[f"{s}_pct"] = (pivot[s] / pivot["total_min"] * 100).round(2)

return pivot.reset_index()

def bottleneck_score(self, summary_df: pd.DataFrame) -> pd.DataFrame:

"""计算瓶颈得分(待机+夹具调整占比越高=越瓶颈)"""

df = summary_df.copy()

df["aux_pct"] = df.get("待机_pct", 0) + df.get("夹具调整_pct", 0)

df["bottleneck_rank"] = df["aux_pct"].rank(ascending=False)

return df.sort_values("aux_pct", ascending=False).reset_index(drop=True)

</details>

<details>

<summary></summary>

"""状态转移分析"""

import numpy as np

import pandas as pd

from typing import Optional

class TransitionAnalyzer:

"""统计设备状态转移频次和概率"""

def __init__(self):

self.states = ["加工", "对刀", "夹具调整", "待机"]

def compute_transitions(self, df: pd.DataFrame) -> pd.DataFrame:

"""计算转移矩阵"""

records = []

for dev, grp in df.sort_values("timestamp").groupby("device_id"):

states = grp["state"].tolist()

for i in range(len(states) - 1):

records.append({

"device_id": dev,

"from_state": states[i],

"to_state": states[i + 1],

})

return pd.DataFrame(records)

def transition_matrix(self, df: pd.DataFrame) -> pd.DataFrame:

"""转移频次矩阵"""

trans = self.compute_transitions(df)

matrix = trans.groupby(["from_state", "to_state"]).size().reset_index(name="count")

pivot = matrix.pivot(index="from_state", columns="to_state", values="count").fillna(0)

for s in self.states:

if s not in pivot.index:

pivot.loc[s] = 0

if s not in pivot.columns:

pivot[s] = 0

pivot = pivot.reindex(self.states, columns=self.states, fill_value=0)

return pivot.astype(int)

def transition_probability(self, matrix: pd.DataFrame) -> pd.DataFrame:

"""转移概率(行归一化)"""

prob = matrix.copy().astype(float)

for idx in prob.index:

row_sum = prob.loc[idx].sum()

if row_sum > 0:

prob.loc[idx] = prob.loc[idx] / row_sum

return prob.round(4)

def device_transition_freq(self, df: pd.DataFrame) -> pd.DataFrame:

"""每台设备的转移频率"""

trans = self.compute_transitions(df)

freq = trans.groupby(["device_id", "from_state", "to_state"]).size().reset_index(name="count")

return freq.sort_values("count", ascending=False).reset_index(drop=True)

</details>

<details>

<summary></summary>

"""状态统计检验 (scipy)"""

import numpy as np

import pandas as pd

from scipy import stats

from typing import Dict, Optional

class StateStatistics:

"""设备间状态占比差异检验"""

def __init__(self, alpha: float = 0.05):

self.alpha = alpha

def machining_ratio_test(self, summary_df: pd.DataFrame) -> Dict:

"""设备间加工占比 ANOVA"""

ratios = []

for _, r in summary_df.iterrows():

dev_ratios = []

for s in ["加工", "对刀", "夹具调整", "待机"]:

col = f"{s}_pct"

if col in r and r[col] > 0:

dev_ratios.append(r[col])

if dev_ratios:

ratios.append(dev_ratios)

if len(ratios) < 2:

return {"f_statistic": np.nan, "p_value": np.nan}

# 用加工占比做ANOVA

groups = []

for _, r in summary_df.iterrows():

if "加工_pct" in r and r["加工_pct"] > 0:

groups.append([r["加工_pct"]])

# 简化: 直接用加工占比列

machining_pcts = summary_df.get("加工_pct", pd.Series()).dropna().values

if len(machining_pcts) < 2:

return {"f_statistic": np.nan, "p_value": np.nan}

# 单样本正态性

if len(machining_pcts) >= 3:

stat, p = stats.shapiro(machining_pcts)

else:

stat, p = np.nan, np.nan

return {

"shapiro_stat": round(float(stat), 4) if not np.isnan(stat) else np.nan,

"shapiro_p": round(float(p), 4) if not np.isnan(p) else np.nan,

"mean_machining_pct": round(float(np.mean(machining_pcts)), 2),

"std_machining_pct": round(float(np.std(machining_pcts, ddof=1)), 2),

}

def device_pairwise_ttest(self, df: pd.DataFrame,

device_a: str, device_b: str,

state: str = "加工") -> Dict:

"""两设备某状态时长 t 检验"""

a_dur = df[(df["device_id"] == device_a) & (df["state"] == state)]["duration_min"].values

b_dur = df[(df["device_id"] == device_b) & (df["state"] == state)]["duration_min"].values

if len(a_dur) < 2 or len(b_dur) < 2:

return {"statistic": np.nan, "p_value": np.nan}

stat, p = stats.ttest_ind(a_dur, b_dur, equal_var=False)

return {

"device_a": device_a, "device_b": device_b,

"state": state,

"mean_a": round(float(np.mean(a_dur)), 2),

"mean_b": round(float(np.mean(b_dur)), 2),

"statistic": round(float(stat), 4),

"p_value": round(float(p), 4),

"significant": bool(p < self.alpha),

}

</details>

<details>

<summary></summary>

"""设备状态画像聚类 (scikit-learn)"""

import numpy as np

import pandas as pd

from sklearn.cluster import KMeans

from sklearn.preprocessing import StandardScaler

from typing import Optional

class DeviceProfiler:

"""基于状态占比的设备聚类"""

def __init__(self, n_clusters: int = 3, random_state: int = 42):

self.n_clusters = n_clusters

self.random_state = random_state

def profile(self, summary_df: pd.DataFrame) -> pd.DataFrame:

"""聚类并标注画像类型"""

df = summary_df.copy()

feat_cols = []

for s in ["加工", "对刀", "夹具调整", "待机"]:

pct_col = f"{s}_pct"

if pct_col in df.columns:

feat_cols.append(pct_col)

else:

df[pct_col] = 0.0

feat_cols.append(pct_col)

X = df[feat_cols].fillna(0).values

scaler = StandardScaler()

Xs = scaler.fit_transform(X)

km = KMeans(n_clusters=self.n_clusters, random_state=self.random_state)

df["profile_cluster"] = km.fit_predict(Xs)

# 标注画像

centers = scaler.inverse_transform(km.cluster_centers_)

labels = {}

for i, center in enumerate(centers):

max_idx = np.argmax(center)

dominant = feat_cols[max_idx].replace("_pct", "")

machining_pct = center[feat_cols.index("加工_pct")] if "加工_pct" in feat_cols else 0

if machining_pct > 50:

labels[i] = f"高效型(加工{machining_pct:.0f}%)"

elif dominant in ("待机", "夹具调整"):

labels[i] = f"瓶颈型(辅助高)"

else:

labels[i] = f"均衡型"

df["profile_label"] = df["profile_cluster"].map(labels)

return df

def cluster_summary(self, df: pd.DataFrame) -> pd.DataFrame:

"""聚类汇总"""

if "profile_label" not in df.columns:

return pd.DataFrame()

rows = []

for lbl, grp in df.groupby("profile_label"):

rows.append({

"profile": lbl,

"device_count": len(grp),

"avg_machining_pct": round(grp.get("加工_pct", 0).mean(), 2),

"avg_aux_pct": round((grp.get("待机_pct", 0) + grp.get("夹具调整_pct", 0)).mean(), 2),

"devices": ", ".join(grp["device_id"].tolist()),

})

return pd.DataFrame(rows).sort_values("avg_machining_pct", ascending=False).reset_index(drop=True)

</details>

<details>

<summary></summary>

"""设备-状态关系网 (networkx)"""

import networkx as nx

import pandas as pd

from typing import Optional

class StateGraph:

"""建设备-状态转移关系网"""

def __init__(self):

self.G = nx.DiGraph()

def build(self, df: pd.DataFrame,

transition_df: Optional[pd.DataFrame] = None) -> nx.DiGraph:

self.G.clear()

# 设备节点

for dev in df["device_id"].unique():

self.G.add_node(f"D:{dev}", ntype="device")

# 状态节点

for s in ["加工", "对刀", "夹具调整", "待机"]:

self.G.add_node(f"S:{s}", ntype="state")

# 边: 设备→状态(按总时长)

for dev, grp in df.groupby("device_id"):

state_dur = grp.groupby("state")["duration_min"].sum() if "duration_min" in grp.columns else grp.groupby("state").size()

for s, dur in state_dur.items():

self.G.add_edge(f"D:{dev}", f"S:{s}",

weight=round(float(dur), 2))

# 状态转移边

if transition_df is not None:

for _, r in transition_df.iterrows():

self.G.add_edge(f"S:{r['from_state']}",

f"S:{r['to_state']}",

weight=float(r["count"]))

return self.G

def bottleneck_devices(self) -> pd.DataFrame:

"""按入度中心性找瓶颈设备"""

if not self.G.nodes:

return pd.DataFrame()

in_cent = nx.in_degree_centrality(self.G)

rows = []

for n, c in in_cent.items():

if self.G.nodes[n].get("ntype") == "device":

rows.append({

"device": n.split(":")[-1],

"centrality": round(c, 4),

})

df = pd.DataFrame(rows)

if df.empty:

return df

return df.sort_values("centrality", ascending=False).reset_index(drop=True)

</details>

<details>

<summary></summary>

"""可视化 (matplotlib)"""

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

from pathlib import Path

plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]

plt.rcParams["axes.unicode_minus"] = False

class StateVisualizer:

def __init__(self, results_dir: str = "results"):

self.results_dir = Path(results_dir)

self.results_dir.mkdir(exist_ok=True)

def stacked_bar(self, summary_df):

"""设备状态堆叠条形图"""

fig, ax = plt.subplots(figsize=(12, 6))

devs = summary_df["device_id"].tolist()

bottoms = np.zeros(len(devs))

colors = {"加工": "#27AE60", "对刀": "#F39C12",

"夹具调整": "#E74C3C", "待机": "#95A5A6"}

for s, c in colors.items():

pct_col = f"{s}_pct"

vals = summary_df[pct_col].values if pct_col in summary_df.columns else np.zeros(len(devs))

ax.bar(devs, vals, bottom=bottoms, color=c, label=s, edgecolor="white", width=0.6)

bottoms += vals

ax.set_ylabel("占比 (%)")

ax.set_title("设备状态时长占比堆叠图", fontsize=14, fontweight="bold")

ax.legend(loc="upper right")

ax.set_ylim(0, 100)

plt.xticks(rotation=45)

plt.tight_layout()

plt.savefig(self.results_dir / "state_stacked_bar.png",

dpi=150, bbox_inches="tight")

plt.close()

def gantt_timeline(self, df, max_points=200):

"""甘特时间线(简化版: 按时间顺序着色)"""

fig, ax = plt.subplots(figsize=(14, 5))

devs = df["device_id"].unique()

dev_map = {d: i for i, d in enumerate(devs)}

colors = {"加工": "#27AE60", "对刀": "#F39C12",

"夹具调整": "#E74C3C", "待机": "#95A5A6"}

df2 = df.sort_values("timestamp")

if len(df2) > max_points:

step = len(df2) // max_points

df2 = df2.iloc[::step]

for _, r in df2.iterrows():

y = dev_map.get(r["device_id"], 0)

ax.scatter(r["timestamp"], y,

color=colors.get(r["state"], "#999"),