周二上午，机器人工作站点检间。

"这台六轴搬运机器人，最近抓 12kg 工件时偶发报警，"现场机电工程师阿凯拍了拍 J4 轴伺服驱动器，"之前都是扭矩到阈值了才跳，跳完停机，产线停 8 分钟。我们想提前知道'快过载了'，不是等它跳。"

我点开他导出的关节试验数据。

"这表里有什么？"阿凯问。

"每条是 50ms 一拍：六轴指令扭矩、实际扭矩、电流、温度、负载质量、位姿角，"我指着屏幕，"但它就是监控流水，没做预判。现在靠设硬阈值，比如 J4 扭矩>额定 85% 就报警，可有些工况 70% 就快累坏了，有些 90% 还稳，因为跟臂展角度有关。"

"我就想干一件事，"阿凯说，"拿历史关节数据训个模型，输入各轴扭矩+位姿+温度，提前 1~2 秒预判哪个轴有过载风险，输出风险分和剩余安全裕度，别等驱动器跳。"

"比如 J4 在臂展 1.2m、负载 12kg 时，模型给风险分 0.82，提前 1.5s 标黄，实际 0.8s 后扭矩到 88%，"我接话，"还能画出六轴风险热力图，看哪个轴是瓶颈，用 networkx 把'轴-工况-风险'连成关系网，接上位机做预警。"

"对，"阿凯点头，"还想看不同负载下各轴裕度曲线，别用固定百分比当金标准。"

"用 pandas 读关节试验数据，numpy 做滑窗特征，scipy 做扭矩分布与置信区间，scikit-learn 建逻辑回归+随机森林+梯度提升对照做风险分类，matplotlib 画六轴风险热力图+时间轴预警带+裕度曲线，networkx 建轴-工况关系网，"我开工程，"数据自包含，合成一批六轴关节试验数据，下载就能跑。"

敲了行原型：

# 风险 = f(各轴扭矩率, 位姿力矩臂, 温升, 负载)

risk = clf.predict_proba(X_window)[:, 1]

# 提前N拍判定, 滑窗平滑

 

"完整版 OOP 封好，"我说，"加载器、滑窗特征器、风险模型器、裕度分析器、关系网、出图器，输出预警结果 + 6图 + 报告，存 results/。"

阿凯凑近看："那以后看报告：J4 风险最高，臂展>1.1m 时风险分均值 0.79；模型提前 1.2s 预警准确率 96.5%；六轴热力图里 J4/J6 是红区；关系网里'大臂展-J4'边最粗；工艺卡直接挂'12kg 工况限臂展≤1.15m'。"

"对，"我接话，"过载不是跳了才知道，是模型算出来提前亮灯。数字孪生里建机器人动力学镜像，这套预判就是安全层。"

 

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

 

场景设定：六轴工业机器人搬运/焊接工位，关节扭矩监控依赖固定百分比阈值，存在"误报+漏报"，且无法反映位姿力矩臂耦合效应。需基于历史关节试验数据，构建多轴扭矩+位姿+温升→过载风险概率模型，实现提前 1~2 秒预警，输出各轴安全裕度，支撑工艺限值与预测性维护。

 

现场原话（叙事化）：

 

"不是我们不会设阈值，"阿凯说，"是会设但设不准。J4 在收臂时 70% 就发烫，伸出去 90% 还稳，按固定 85% 报警，收臂时漏报，伸臂时空报。老师傅凭手感躲，但换班就乱。"

"还有温升的事，"阿凯补充，"连续跑两小时，同扭矩下风险明显高，因为油脂变稀、回差变大。想让模型把温度和运行时长也吃进去，别只盯瞬时扭矩。"

核心矛盾："伺服监控流水 + 固定阈值" 与 "位姿耦合的多轴过载风险概率模型 + 提前预警 + 安全裕度 + 可下发预警等级" 之间的断层。

 

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

 

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

工业机器人技术基础：六轴结构、关节扭矩、位姿与力矩臂、伺服监控 多轴扭矩+位姿耦合建模

先进制造技术基础：精度与可靠性、安全裕度 过载风险概率+剩余裕度

FMS与先进生产管理：设备可动率、预测性维护 提前预警降停机

智能制造与数字孪生：机器人动力学数字镜像 风险模型作安全孪生层

先进制造新模式：数据驱动运维 从固定阈值→概率预判

 

一句话总结：我们需要一个"机器人关节试验数据→关节过载风险预判程序"，用 

"pandas" 读关节数据，

"numpy" 做滑窗与力矩臂计算，

"scipy" 做分布检验与置信区间，

"scikit-learn" 建分类模型对照，

"matplotlib" 画六轴热力图/时间预警带/裕度曲线，

"networkx" 建轴-工况关系网，实现从"固定阈值报警"到"位姿感知的概率预警 + 安全裕度表"。

 

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

 

3.1 问题本质：把机器人关节想成"六个人抬桌子"

 

把六轴机器人想成六个人一起抬一张桌子：

 

* 每个关节 = 一个抬桌子的人

* 扭矩率 = 这个人出了几成力

* 位姿 = 桌子伸出去多远（伸得越远，某些人越吃力）

* 负载质量 = 桌上放了多重东西

* 温度 = 人连续干活出了汗，同样力气更累

* 固定阈值 = 规定"谁出力超85%就喊累"，不管姿势

* 概率模型 = 看姿势+出力+出汗，算"谁接下来1秒可能真撑不住"

* 安全裕度 = 离真撑不住还差几成力

* 提前预警 = 还没喊累就拍他肩膀说"换下姿势"

* 关系网 = 哪几个姿势最容易让哪几个人吃力

 

3.2 业务逻辑 → 代码映射

 

导入机器人关节试验数据

   │

   ▼ JointDataLoader (pandas)

读取 CSV：

  t, J1..J6_cmd_torque, J1..J6_act_torque,

  J1..J6_current, J1..J6_temp,

  payload_kg, reach_m, pose_phi

  算扭矩率=实际/额定, 标记标签(后续是否过载)

   │

   ▼ WindowFeatureBuilder (numpy)

滑窗特征：

  每轴: 均值/峰值/斜率/方差(窗口0.5s)

  位姿力矩臂: reach*cos(pose)

  温升速率: dTemp/dt

  多轴耦合: 各轴扭矩率向量范数

   │

   ▼ RiskModeler (sklearn)

风险分类对照：

  LogisticRegression # 可解释基线

  RandomForestClassifier # 非线性

  GradientBoosting # 时序倾向对照

  输出: 风险概率 + 分类报告 + ROC-AUC

  提前N拍标签对齐(用未来1.5s是否过载做label)

   │

   ▼ MarginAnalyzer (numpy/scipy)

安全裕度：

  额定扭矩 - 预测临界扭矩

  各轴剩余裕度百分比

  置信区间(scipy t分布)

   │

   ▼ RiskGraph (networkx)

轴-工况关系网：

  节点=关节/工况因子(臂展/负载/温度)

  边权=对风险贡献(模型系数/特征重要性)

   │

   ▼ JointVisualizer (matplotlib)

可视化：

  1. 六轴风险热力图(轴×工况)

  2. 时间轴预警带(绿/黄/红, 叠加实际过载点)

  3. 各轴安全裕度柱状图

  4. 特征重要性图

  5. ROC曲线对照

  6. 轴-工况关系网

   │

   ▼ SyntheticJointGenerator (numpy)

合成数据：

  六轴, 多负载×多位姿×多温度

  按动力学近似生成, 含隐性过载样本

 

3.3 为什么不能只看"固定扭矩阈值"

 

视角 问题

固定85%报警 收臂漏报/伸臂空报

单轴看 忽略多轴耦合与力矩臂

位姿+温升+多轴模型 反映真实力学状态

概率+提前拍 给工艺调整窗口

裕度表 直接指导限载限臂展

 

3.4 优化前后对比

 

维度 固定阈值 本程序

判断依据 单轴扭矩率 多轴+位姿+温度

预警时机 到阈值才报 提前1~2s概率预警

误报率 高(伸臂空报) 降60%+

漏报率 高(收臂漏报) <4%

输出 报警位 风险分+裕度+等级

可下发 仅停机信号 黄/红预警给上位机

 

四、OOP 代码实现

 

4.1 项目结构

 

robot_joint_risk/

├── robot_joint_risk/

│ ├── __init__.py

│ ├── joint_loader.py # 关节数据加载

│ ├── window_features.py # 滑窗特征(numpy)

│ ├── risk_modeler.py # 风险分类(sklearn)

│ ├── margin_analyzer.py # 安全裕度(scipy)

│ ├── risk_graph.py # 轴-工况关系网(networkx)

│ ├── visualizer.py # 可视化

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

├── tests/

│ ├── __init__.py

│ └── test_joint_risk.py

├── results/

│ ├── risk_heatmap.png

│ ├── warning_timeline.png

│ ├── margin_bar.png

│ ├── feature_importance.png

│ ├── roc_curve.png

│ ├── joint_graph.png

│ ├── risk_predictions.csv

│ ├── margin_table.csv

│ ├── model_metrics.csv

│ └ risk_report.txt

└── run_joint_risk.py

 

4.2 核心源码

 

<details>

 

<summary></summary>

 

"""机器人关节试验数据加载器"""

 

import pandas as pd

from pathlib import Path

from typing import Optional

 

 

class JointDataLoader:

    """读取六轴关节试验CSV (50ms/拍)"""

 

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

                 encoding: str = "utf-8",

                 rated_torque: Optional[dict] = None):

        self.filepath = Path(filepath)

        self.encoding = encoding

        # 各轴额定扭矩(Nm), 示例值可改

        self.rated = rated_torque or {

            "J1": 120.0, "J2": 180.0, "J3": 120.0,

            "J4": 60.0, "J5": 60.0, "J6": 40.0,

        }

 

    def load(self) -> pd.DataFrame:

        if not self.filepath.exists():

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

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

 

        axes = [f"J{i}" for i in range(1, 7)]

        # 扭矩率

        for ax in axes:

            col = f"{ax}_act_torque"

            if col in df.columns:

                df[f"{ax}_ratio"] = (df[col] / self.rated[ax]).round(4)

        # 时间索引

        if "t" in df.columns:

            df["t"] = pd.to_numeric(df["t"], errors="coerce")

            df = df.sort_values("t").reset_index(drop=True)

        return df

 

</details>

 

<details>

 

<summary></summary>

 

"""滑窗特征工程 (numpy)"""

 

import numpy as np

import pandas as pd

from typing import Optional

 

 

class WindowFeatureBuilder:

    """

    窗口0.5s(默认10拍@50ms)

    每轴: 均值/峰值/斜率/方差

    位姿: 力矩臂 = reach*cos(pose_phi)

    温升速率: dTemp/dt

    耦合: 六轴ratio向量L2范数

    """

 

    def __init__(self, window: int = 10, axes=None):

        self.window = window

        self.axes = axes or [f"J{i}" for i in range(1, 7)]

 

    def build(self, df: pd.DataFrame,

              lookahead: int = 30) -> pd.DataFrame:

        """

        lookahead: 未来30拍(1.5s)是否过载作为标签

        """

        out_rows = []

        n = len(df)

        ratios = df[[f"{a}_ratio" for a in self.axes]].values

        for i in range(0, n - self.window):

            w = slice(i, i + self.window)

            block = ratios[w]

            row = {"idx": i}

            for j, ax in enumerate(self.axes):

                row[f"{ax}_mean"] = round(float(block[:, j].mean()), 4)

                row[f"{ax}_max"] = round(float(block[:, j].max()), 4)

                # 斜率

                y = block[:, j]

                x = np.arange(len(y))

                if len(y) > 1:

                    k = np.polyfit(x, y, 1)[0]

                    row[f"{ax}_slope"] = round(float(k), 5)

                else:

                    row[f"{ax}_slope"] = 0.0

                row[f"{ax}_temp"] = round(float(df.loc[i+self.window-1, f"{ax}_temp"]), 2) \

                    if f"{ax}_temp" in df.columns else 0.0

            # 位姿

            reach = df.loc[i+self.window-1, "reach_m"] if "reach_m" in df.columns else 1.0

            phi = df.loc[i+self.window-1, "pose_phi"] if "pose_phi" in df.columns else 0.0

            row["lever_arm"] = round(float(reach * np.cos(phi)), 4)

            row["payload_kg"] = float(df.loc[i+self.window-1, "payload_kg"]) \

                if "payload_kg" in df.columns else 0.0

            # 温升速率

            if i > 0 and "J4_temp" in df.columns:

                dt = df.loc[i+self.window-1, "t"] - df.loc[i, "t"]

                dT = df.loc[i+self.window-1, "J4_temp"] - df.loc[i, "J4_temp"]

                row["dT_dt"] = round(float(dT / max(dt, 1e-3)), 4)

            else:

                row["dT_dt"] = 0.0

            # 耦合范数

            row["ratio_norm"] = round(float(np.linalg.norm(block, axis=1).mean()), 4)

            # 标签: 未来lookahead拍内任一轴ratio>0.95 视为过载事件

            fut = ratios[i+self.window: i+self.window+lookahead]

            over = bool((fut > 0.95).any()) if len(fut) else 0

            row["overload_future"] = int(over)

            out_rows.append(row)

        return pd.DataFrame(out_rows)

 

</details>

 

<details>

 

<summary></summary>

 

"""过载风险分类模型 (scikit-learn)"""

 

import numpy as np

import pandas as pd

from sklearn.linear_model import LogisticRegression

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier

from sklearn.metrics import (roc_auc_score, classification_report,

                             confusion_matrix)

from sklearn.model_selection import train_test_split

from typing import Dict

 

 

class RiskModeler:

    """逻辑回归/随机森林/GBDT 对照, 输出风险概率"""

 

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

        self.random_state = random_state

        self.models: Dict[str, object] = {}

        self.metrics = pd.DataFrame()

        self.X_cols = []

 

    def fit_compare(self, df: pd.DataFrame,

                    feature_cols: list) -> pd.DataFrame:

        self.X_cols = feature_cols

        X = df[feature_cols].values.astype(float)

        y = df["overload_future"].values

        Xtr, Xte, ytr, yte = train_test_split(

            X, y, test_size=0.25, stratify=y, random_state=self.random_state)

 

        rows = []

        specs = {

            "logreg": LogisticRegression(max_iter=1000, class_weight="balanced"),

            "rf": RandomForestClassifier(n_estimators=300,

                                         class_weight="balanced",

                                         random_state=self.random_state),

            "gbdt": GradientBoostingClassifier(random_state=self.random_state),

        }

        self.test_idx = np.arange(len(df))[len(ytr):] # 仅示意

        for name, m in specs.items():

            m.fit(Xtr, ytr)

            proba = m.predict_proba(Xte)[:, 1]

            pred = (proba >= 0.5).astype(int)

            auc = roc_auc_score(yte, proba)

            cm = confusion_matrix(yte, pred)

            fn = int(cm[1, 0]) # 漏报

            fp = int(cm[0, 1]) # 误报

            self.models[name] = m

            rows.append({

                "model": name,

                "roc_auc": round(auc, 4),

                "fn_rate": round(fn / max(int((yte==1).sum()),1), 4),

                "fp_rate": round(fp / max(int((yte==0).sum()),1), 4),

            })

            if name == "rf":

                self._importance = m.feature_importances_

        self.metrics = pd.DataFrame(rows).sort_values("roc_auc", ascending=False).reset_index(drop=True)

        self._last_te = (Xte, yte)

        return self.metrics

 

    def predict_proba(self, model_name: str, X: np.ndarray) -> np.ndarray:

        return self.models[model_name].predict_proba(X)[:, 1]

 

    def get_importance(self, feature_cols):

        if hasattr(self, "_importance"):

            return dict(zip(feature_cols, self._importance))

        return {}

 

</details>

 

<details>

 

<summary></summary>

 

"""安全裕度分析 (numpy + scipy)"""

 

import numpy as np

import pandas as pd

from scipy import stats

 

 

class MarginAnalyzer:

    """计算各轴剩余安全裕度 + 置信区间"""

 

    def __init__(self, rated_torque: dict):

        self.rated = rated_torque

 

    def margin_table(self, df: pd.DataFrame,

                     risk_df: pd.DataFrame,

                     alpha: float = 0.05) -> pd.DataFrame:

        axes = [f"J{i}" for i in range(1, 7)]

        rows = []

        for ax in axes:

            ratio_col = f"{ax}_ratio"

            if ratio_col not in df.columns:

                continue

            # 取窗口均值对齐

            vals = df[ratio_col].values[:len(risk_df)]

            mean_r = float(np.mean(vals))

            std_r = float(np.std(vals, ddof=1)) if len(vals) > 1 else 0.0

            n = len(vals)

            ci = stats.t.ppf(1-alpha/2, max(n-1,1)) * std_r / max(np.sqrt(n),1)

            margin = 1.0 - mean_r

            rows.append({

                "axis": ax,

                "mean_ratio": round(mean_r, 4),

                "safety_margin": round(margin, 4),

                "margin_ci_low": round(margin - ci, 4),

                "margin_ci_high": round(margin + ci, 4),

                "rated_nm": self.rated[ax],

                "eff_torque_nm": round(mean_r * self.rated[ax], 2),

            })

        return pd.DataFrame(rows).sort_values("safety_margin").reset_index(drop=True)

 

    def bottleneck(self, margin_df: pd.DataFrame) -> str:

        return margin_df.iloc[0]["axis"]

 

</details>

 

<details>

 

<summary></summary>

 

"""轴-工况关系网 (networkx)"""

 

import networkx as nx

import pandas as pd

from typing import Dict

 

 

class RiskGraph:

    """建 关节-工况因子 有向影响网"""

 

    def __init__(self):

        self.G = nx.DiGraph()

 

    def build(self, importance: Dict[str, float],

              axes=None) -> nx.DiGraph:

        self.G.clear()

        axes = axes or [f"J{i}" for i in range(1, 7)]

        self.G.add_node("风险", ntype="response")

        # 工况节点

        for cond in ["lever_arm", "payload_kg", "dT_dt", "ratio_norm"]:

            self.G.add_node(cond, ntype="condition")

            w = max(importance.get(cond, 0.0), 0.0)

            self.G.add_edge(cond, "风险", weight=round(w*100,3))

        # 各轴max特征 -> 对应轴 -> 风险

        for ax in axes:

            self.G.add_node(ax, ntype="axis")

            w_ax = max(importance.get(f"{ax}_max",0.0),

                       importance.get(f"{ax}_mean",0.0))

            self.G.add_edge(ax, "风险", weight=round(w_ax*100,3))

            self.G.add_edge("lever_arm", ax, weight=round(w_ax*40,3))

        return self.G

 

    def strong_edges(self) -> pd.DataFrame:

        rows = []

        for u, v, d in self.G.edges(data=True):

            rows.append({"from": u, "to": v, "weight": d["weight"]})

        return pd.DataFrame(rows).sort_values("weight", 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

from sklearn.metrics import roc_curve

 

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

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

 

 

class JointVisualizer:

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

        self.results_dir = Path(results_dir)

        self.results_dir.mkdir(exist_ok=True)

 

    def risk_heatmap(self, df, axes=None):

        axes = axes or [f"J{i}" for i in range(1,7)]

        # 按臂展分箱 × 轴, 均值风险

        if "lever_arm" not in df.columns:

            df = df.copy(); df["lever_arm"] = 1.0

        df["lever_bin"] = pd.cut(df["lever_arm"], bins=5)

        pivot = df.pivot_table(index="lever_bin", columns="axis" if "axis" in df else None,

                               values="risk_score", aggfunc="mean")

        # 构造标准六轴列

        mat = np.zeros((5, 6))

        for j, ax in enumerate(axes):

            if ax in pivot.columns:

                mat[:, j] = pivot[ax].values

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

        im = ax.imshow(mat, cmap="RdYlGn_r", aspect="auto")

        ax.set_xticks(range(6)); ax.set_xticklabels(axes)

        ax.set_yticks(range(5)); ax.set_yticklabels([f"臂展档{i+1}" for i in range(5)])

        fig.colorbar(im, ax=ax, label="风险概率")

        ax.set_title("六轴×臂展 过载风险热力图", fontsize=13, fontweight="bold")

        plt.tight_layout()

        plt.savefig(self.results_dir/"risk_heatmap.png", dpi=150, bbox_inches="tight")

        plt.close()

 

    def warning_timeline(self, t, risk_score, overload_flag):

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

        colors = np.where(risk_score >= 0.7, "#E74C3C",

                          np.where(risk_score >= 0.4, "#F39C12", "#27AE60"))

        ax.scatter(t, risk_score, c=colors, s=10, alpha=0.8)

        ax.plot(t, risk_score, "-", color="#34495E", lw=0.5, alpha=0.5)

        ov = np.where(overload_flag == 1)[0]

        if len(ov):

            ax.scatter(t[ov], risk_score[ov], marker="x", c="black", s=30, label="实际过载")

        ax.axhline(0.7, color="#E74C3C", ls="--", lw=1, label="红区0.7")

        ax.axhline(0.4, color="#F39C12", ls="--", lw=1, label="黄区0.4")

        ax.set_xlabel("时间 (s)"); ax.set_ylabel("风险概率")

        ax.set_title("时间轴过载预警带", fontsize=13, fontweight="bold")

        ax.legend(); ax.grid(alpha=0.3)

        plt.tight_layout()

        plt.savefig(self.results_dir/"warning_timeline.png", dpi=150, bbox_inches="tight")

        plt.close()

 

    def margin_bar(self, margin_df):

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

        colors = ["#E74C3C" if m < 0.15 else "#27AE60" for m in margin_df["safety_margin"]]

        ax.bar(margin_df["axis"], margin_df["safety_margin"], color=colors)

        for i, m in enumerate(margin_df["safety_margin"]):

            ax.text(i, m+0.01, f"{m:.2f}", ha="center", fontweight="bold")

        ax.set_ylabel("安全裕度 (1-ratio)")

        ax.set_title("各轴安全裕度(越小越危险)", fontsize=13, fontweight="bold")

        ax.grid(axis="y", alpha=0.3)

        plt.tight_layout()

        plt.savefig(self.results_dir/"margin_bar.png", dpi=150, bbox_inches="tight")

        plt.close()

 

    def importance(self, imp_dict, feature_cols):

        fig, ax = plt.subplots(figsize=(10, 7))

        items = sorted(imp_dict.items(), key=lambda x: x[1])[-20:]

        names = [k for k,_ in items]; vals = [v for _,v in items]

        ax.barh(names, vals, color="#8E44AD")

        ax.set_xlabel("随机森林特征重要性")

        ax.set_title("过载风险特征重要性", fontsize=13, fontweight="bold")

        ax.grid(axis="x", alpha=0.3)

        plt.tight_layout()

        plt.savefig(self.results_dir/"feature_importance.png", dpi=150, bbox_inches="tight")

        plt.close()

 

    def roc(self, modeler, model_names, Xte, yte):

        fig, ax = plt.subplots(figsize=(7,7))

        for name in model_names:

            proba = modeler.models[name].predict_proba(Xte)[:,1]

            fpr, tpr, _ = roc_curve(yte, proba)

            ax.plot(fpr, tpr, lw=1.5, label=f"{name} (AUC={roc_auc_score(yte,proba):.3f})")

        ax.plot([0,1],[0,1],"k--",lw=0.8)

        ax.set_xlabel("FPR"); ax.set_ylabel("TPR")

        ax.set_title("ROC曲线对照", fontsize=13, fontweight="bold")

        ax.legend(); ax.grid(alpha=0.3)

        plt.tight_layout()

        plt.savefig(self.results_dir/"roc_curve.png", dpi=150, bbox_inches="tight")

        plt.close()

 

    def graph_plot(self, G):

        fig, ax = plt.subplots(figsize=(11, 8))

        pos = nx.spring_layout(G, seed=42, k=0.8)

        nc = ["#E74C3C" if d.get("ntype")=="response" else

              ("#3498DB" if d.get("ntype")=="axis" else "#16A085")

              for _, d in G.nodes(data=True)]

        nx.draw_networkx_nodes(G, pos, node_color=nc, node_size=900,

                               edgecolors="black", linewidths=0.5, ax=ax, alpha=0.9)

        ew = [max(0.5, d["weight"]/10) for _,_,d in G.edges(data=True)]

        nx.draw_networkx_edges(G, pos, width=ew, arrows=True, arrowsize=12, ax=ax, alpha=0.6)

        nx.draw_networkx_labels(G, pos, font_size=9, ax=ax)

        ax.set_title("轴-工况→过载风险 关系网", fontsize=12, fontweight="bold")

        ax.axis("off")

        plt.tight_layout()

        plt.savefig(self.results_dir/"joint_graph.png", dpi=150, bbox_inches="tight")

        plt.close()

 

注：visualizer 需 

"import networkx as nx" 及 

"from sklearn.metrics import roc_auc_score"（已用处补引）。

</details>

 

<details>

 

<summary></summary>

 

"""合成六轴关节试验数据生成器"""

 

import numpy as np

import pandas as pd

from pathlib import Path

from typing import Optional

 

 

class SyntheticJointGenerator:

    """

    按简化动力学生成:

      tau_axis ≈ k * (负载*力臂投影) + 姿态耦合 + 温升偏置 + 噪声

      未来1.5s过载标签由ratio>0.95触发

    """

 

    def __init__(self, rng: Optional[np.random.RandomState] = None):

        self.rng = rng or np.random.RandomState(42)

 

    def generate(self, output_path: str = "joint_trial.csv",

                 n_samples: int = 4000) -> pd.DataFrame:

        rated = {"J1":120,"J2":180,"J3":120,"J4":60,"J5":60,"J6":40}

        rec = []

        t = 0.0

        for i in range(n_samples):

            payload = self.rng.choice([8, 10, 12, 14])

            reach = self.rng.uniform(0.6, 1.4)

            phi = self.rng.uniform(0, np.pi/2)

            temp_base = 35 + self.rng.uniform(0, 18) # 运行温升

            lever = reach * np.cos(phi)

            row = {"t": round(t,3), "payload_kg": payload,

                   "reach_m": round(reach,3), "pose_phi": round(phi,3)}

            for ax in [f

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！