周三下午,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"),

                 

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐