周二下午，焊接车间机器人工作站。

"这批法兰环缝，焊完探伤发现 6 道未熔合，"焊接工艺员小周指着 X 光底片，"都在起弧后 3-8 秒的位置，你肉眼看焊缝表面好好的，打磨了才看到根部的未熔合。返修一次要切掉重焊，工时翻倍。"

我调出机器人焊接数据库的 CSV 导出文件。

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

"时间戳、焊接电流(A)、焊接电压(V)、送丝速度(m/min)、保护气流量(L/min)、焊枪角度(°)、工件间隙(mm)、层间温度(℃)、焊接模式（短路/脉冲/喷射），以及对应的探伤结果（合格/未熔合/气孔/咬边），"我指着屏幕，"但它就是一张逐点记录的工艺参数表。你能看到未熔合那几道的电流波形在起弧阶段有个 15A 左右的塌陷，持续了 0.4 秒，可你没法逐道人工看波形——一天焊 200 道，谁看得过来？"

"我就想干一件事，"小周说，"给我一个程序：实时读焊接电流波形，在焊完 5 秒内告诉我'这道焊缝有没有未熔合风险'。不用 100% 准确，但能把高风险的那几道挑出来，优先去探伤，低风险的直接放行。这样探伤工作量砍一半，返修也从'事后发现'变成'焊完就知道'。"

"比如起弧阶段电流偏低+工件间隙偏大+层间温度偏低，这三个凑一起就容易未熔合，"我接话，"用 pandas 做波形分段特征提取（起弧段/稳态段/收弧段），scipy 做电流波形的统计特征（均值/方差/峭度/下降斜率）+ 短时能量 + 过零率，scikit-learn 训练随机森林分类器预判缺陷，matplotlib 画电流波形叠加缺陷标记+特征分布+ROC曲线+混淆矩阵+特征重要性+决策边界，networkx 建工艺参数→缺陷模式的因果链路。"

"对，"小周点头，"别给我黑盒，要能说清楚'为什么这道被判高风险'。我看得懂，能拿去跟焊工说'起弧电流设到 180A 以上，间隙超过 1.2mm 就先补丝再焊，层温低于 80℃ 就预热'。"

"用 scipy 提取波形时域特征，pandas 做特征汇总，scikit-learn 随机森林+交叉验证，matplotlib 出 6 图+报告，存 results/，"我开工程，"数据自包含，合成一批含 200 道焊缝、未熔合率 15% 的电流波形数据，下载就能跑。"

敲了行原型：

# 未熔合 = 热输入不足 → 电流偏低/电压偏低/速度过快

# 波形特征 = 起弧段能量够不够、稳态段稳不稳

# 分类器 = 学会"什么样的波形形状对应未熔合"

# 时域特征 = 不用频域，电流波形本身就是慢变的

 

"完整版 OOP 封好，"我说，"数据加载器、波形特征提取器、缺陷分类器、因果网络、可视化器，输出缺陷概率+关键特征解释+6图+报告。"

小周凑近看："那以后看报告：随机森林分类 F1=0.89，特征重要性排第一的是'起弧段平均电流'(31%)，第二是'起弧段电流下降斜率'(22%)，第三是'工件间隙'(18%)。决策边界：起弧段平均电流 <165A + 间隙 >1.0mm → 未熔合概率 78%。结论：工艺卡加一条'起弧电流下限 170A，间隙超 1.0mm 必须预填丝'。高风险焊缝焊完自动标记，优先拍片。"

"对，"我接话，"焊接不是'焊上就行'，是'每一道的热输入都要够'。数字孪生里挂焊接质量节点，这套就是焊工的'未熔合预警器'。"

 

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

 

场景设定：机器人 MIG/MAG 焊接工作站，批量焊接法兰、管道环缝、箱体结构件。焊接过程中电流、电压逐毫秒记录，但缺陷检测依赖焊后 X 光/超声波探伤——发现未熔合时工件已焊完，返修成本高（切掉重焊）。

 

现场原话（叙事化）：

 

"不是我们焊不好，"小周说，"是未熔合这东西太隐蔽。表面成型漂亮得很，纹路均匀、余高合适，你看着就是一道好焊缝。结果探伤一照，根部没熔进去——热输入不够，母材和焊丝没真正融合。返修要碳弧气刨切掉，重新坡口、预热、再焊，一道缝返修费是正常焊接的 4 倍。"

"最坑的是起弧位置，"小周补充，"机器人起弧有个建立电弧的过程，前 2-3 秒电流不稳定。如果起弧电流设低了，或者工件间隙偏大，这段就是未熔合的高发区。但你在现场听声音、看弧光，根本分辨不出来。只有探伤能确认。"

核心矛盾："焊后探伤才发现缺陷" 与 "从电流波形实时预判缺陷风险+优先探伤+工艺参数闭环" 之间的断层。

 

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

 

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

工业机器人技术基础：焊接机器人工艺 焊接参数优化+缺陷预防

先进制造技术基础：焊接方法与质量检测 未熔合机理+探伤

智能制造与数字孪生：在线质量监控 焊接过程信号→缺陷预测

柔性制造系统FMS：过程质量控制 实时分选+优先探伤

 

一句话总结：我们需要一个"机器人焊接电流波形→时域特征提取+随机森林分类+缺陷预判+因果网络程序"，用 

"pandas" 做波形分段和特征汇总，

"numpy" 做时域计算，

"scipy" 做统计特征（峭度/偏度/短时能量），

"matplotlib" 画波形叠加标记+ROC+混淆矩阵+特征重要性+决策边界+因果网络，

"scikit-learn" 随机森林分类，

"networkx" 建因果链路，实现从"焊后探伤"到"波形预判+优先探伤+工艺闭环"。

 

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

 

3.1 问题本质：把未熔合想成"煮粥没煮熟"

 

把焊接未熔合想成煮粥没煮熟：

 

* 未熔合 = 米还是硬的，水没烧开就关火了

* 焊接电流 = 火力大小

* 起弧段 = 刚开火那会儿，火还没上来

* 稳态段 = 火稳定了，咕嘟咕嘟

* 工件间隙 = 锅里的米铺得多厚（厚了火要更大）

* 分类器 = 看了 200 锅粥的"火候曲线"，学会了"什么样的火候曲线煮出来是夹生饭"

 

3.2 业务逻辑 → 代码映射

 

读取焊接工艺数据

   │

   ▼ WeldDataLoader (pandas)

读取表:

  时间戳(ms), 电流(A), 电压(V), 送丝速度,

  保护气流量, 焊枪角度, 工件间隙, 层间温度,

  焊接模式, 探伤结果(合格/未熔合/气孔/咬边)

   │

   ▼ WaveformFeatureExtractor (scipy + numpy)

波形分段:

  起弧段(0-3000ms) → 统计特征

  稳态段(3000ms-结束前2000ms) → 统计特征

  收弧段(最后2000ms) → 统计特征

  特征: 均值/方差/最小值/峭度/下降斜率/短时能量

   │

   ▼ DefectClassifier (scikit-learn)

缺陷分类:

  随机森林(主模型) + 逻辑回归(基线)

  5折交叉验证

  输出: 缺陷概率 + 类别标签

   │

   ▼ DefectCausalNetwork (networkx)

因果链路:

  节点: 工艺参数 → 波形特征 → 缺陷类型

  边: 影响方向 + 重要性权重

   │

   ▼ WeldVisualizer (matplotlib)

可视化:

  1. 电流波形叠加缺陷标记(合格=绿, 未熔合=红)

  2. 特征分布箱线图(合格 vs 未熔合)

  3. ROC曲线

  4. 混淆矩阵

  5. 特征重要性柱状图

  6. 因果网络图

   │

   ▼ SyntheticWeldData (numpy)

合成数据:

  200道焊缝, 未熔合率15%

  起弧电流偏低+间隙偏大 → 未熔合

  可复现

 

3.3 为什么不能只看"平均电流"

 

视角 问题

平均电流正常就放行 起弧段塌陷被平均掩盖

焊后全检探伤 成本高、周期长

波形分段特征 抓住起弧段这个关键窗口

分类器 学会"波形形状→缺陷"映射

优先探伤 高风险才拍片，降本增效

 

3.4 分析前后对比

 

维度 传统方式 本程序

缺陷发现 焊后探伤 焊完 5 秒预判

探伤策略 100% 全检 高风险优先，降本 50%

根因分析 "可能电流低了" 起弧段电流<165A + 间隙>1.0mm

工艺闭环 无 预判→调参数→再焊

输出 返修单 6图+报告+缺陷概率

 

四、OOP 代码实现

 

4.1 项目结构

 

weld_defect_predictor/

├── weld_defect_predictor/

│ ├── __init__.py

│ ├── weld_data_loader.py # 数据加载

│ ├── waveform_feature_extractor.py # 波形特征提取

│ ├── defect_classifier.py # 缺陷分类

│ ├── defect_causal_network.py # 因果网络

│ ├── visualizer.py # 可视化

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

├── tests/

│ ├── __init__.py

│ └── test_weld.py

├── results/

│ ├── waveform_overlay.png

│ ├── feature_boxplot.png

│ ├── roc_curve.png

│ ├── confusion_matrix.png

│ ├── feature_importance.png

│ ├── causal_network.png

│ ├── defect_detail.csv

│ └── defect_report.txt

└── run_weld.py

 

4.2 核心源码

 

<details>

 

<summary></summary>

 

"""机器人焊接数据加载器"""

 

import pandas as pd

from pathlib import Path

from typing import Optional

 

 

class WeldDataLoader:

    """读取焊接工艺数据"""

 

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

                 encoding: str = "utf-8"):

        self.filepath = Path(filepath)

        self.encoding = encoding

 

    def load(self) -> pd.DataFrame:

        if not self.filepath.exists():

            raise FileNotFoundError(self.filepath)

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

        req = ["weld_id", "timestamp_ms", "current_a", "voltage_v",

               "wire_feed_m_min", "gas_flow_lpm", "torch_angle_deg",

               "gap_mm", "interpass_temp_c", "weld_mode",

               "defect_type"]

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

        if miss:

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

        num_cols = ["timestamp_ms", "current_a", "voltage_v",

                    "wire_feed_m_min", "gas_flow_lpm", "torch_angle_deg",

                    "gap_mm", "interpass_temp_c"]

        for c in num_cols:

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

        df = df.dropna(subset=["current_a", "defect_type"]).reset_index(drop=True)

        # 二值化: 未熔合 vs 其他(合格+其他缺陷)

        df["is_lack_of_fusion"] = (

            df["defect_type"] == "未熔合"

        ).astype(int)

        return df

 

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

        s = f"总道数: {df['weld_id'].nunique()}\n"

        s += f"总采样点: {len(df)}\n"

        s += f"未熔合率: {df['is_lack_of_fusion'].mean()*100:.1f}%\n"

        s += f"焊接模式: {df['weld_mode'].unique().tolist()}"

        return s

 

</details>

 

<details>

 

<summary></summary>

 

"""焊接电流波形特征提取 (scipy + numpy)"""

 

import numpy as np

import pandas as pd

from scipy import stats

from typing import Dict, List

import warnings

warnings.filterwarnings("ignore")

 

 

class WaveformFeatureExtractor:

    """分段提取电流波形时域特征"""

 

    def __init__(self, arc_start_ms: float = 3000.0,

                 arc_end_ms: float = 2000.0):

        self.arc_start_ms = arc_start_ms

        self.arc_end_ms = arc_end_ms

 

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

        """按 weld_id 分组提取特征"""

        feature_rows = []

        for weld_id, group in df.groupby("weld_id"):

            group = group.sort_values("timestamp_ms").reset_index(drop=True)

            max_t = group["timestamp_ms"].max()

            # 分段

            start_mask = group["timestamp_ms"] <= self.arc_start_ms

            end_mask = group["timestamp_ms"] >= (max_t - self.arc_end_ms)

            steady_mask = (~start_mask) & (~end_mask)

            feat = {"weld_id": weld_id}

            # 全局特征

            feat.update(self._time_domain_features(

                group["current_a"].values, prefix="global"))

            # 起弧段

            if start_mask.sum() > 5:

                feat.update(self._time_domain_features(

                    group.loc[start_mask, "current_a"].values,

                    prefix="start"))

                # 起弧段电流下降斜率(前500ms vs 后500ms)

                start_data = group.loc[start_mask, "current_a"].values

                if len(start_data) >= 10:

                    first_half = start_data[:len(start_data)//2]

                    second_half = start_data[len(start_data)//2:]

                    feat["start_current_drop"] = float(

                        np.mean(first_half) - np.mean(second_half))

                else:

                    feat["start_current_drop"] = 0.0

            else:

                feat.update({f"start_{k}": 0.0 for k in

                           ["mean","std","min","kurtosis","energy"]})

                feat["start_current_drop"] = 0.0

            # 稳态段

            if steady_mask.sum() > 10:

                feat.update(self._time_domain_features(

                    group.loc[steady_mask, "current_a"].values,

                    prefix="steady"))

            else:

                feat.update({f"steady_{k}": 0.0 for k in

                           ["mean","std","min","kurtosis","energy"]})

            # 收弧段

            if end_mask.sum() > 5:

                feat.update(self._time_domain_features(

                    group.loc[end_mask, "current_a"].values,

                    prefix="end"))

            else:

                feat.update({f"end_{k}": 0.0 for k in

                           ["mean","std","min","kurtosis","energy"]})

            # 工艺参数(取均值)

            feat["gap_mm"] = group["gap_mm"].mean()

            feat["interpass_temp_c"] = group["interpass_temp_c"].mean()

            feat["wire_feed_m_min"] = group["wire_feed_m_min"].mean()

            feat["gas_flow_lpm"] = group["gas_flow_lpm"].mean()

            # 标签

            feat["is_lack_of_fusion"] = group["is_lack_of_fusion"].iloc[0]

            feature_rows.append(feat)

        return pd.DataFrame(feature_rows)

 

    def _time_domain_features(self, signal: np.ndarray,

                              prefix: str) -> Dict:

        """时域统计特征"""

        if len(signal) < 3:

            return {f"{prefix}_{k}": 0.0 for k in

                    ["mean","std","min","kurtosis","energy"]}

        return {

            f"{prefix}_mean": round(float(np.mean(signal)), 2),

            f"{prefix}_std": round(float(np.std(signal)), 2),

            f"{prefix}_min": round(float(np.min(signal)), 2),

            f"{prefix}_kurtosis": round(float(stats.kurtosis(signal)), 2),

            f"{prefix}_energy": round(float(np.sum(signal**2)), 2),

        }

 

</details>

 

<details>

 

<summary></summary>

 

"""缺陷分类器 (scikit-learn)"""

 

import numpy as np

import pandas as pd

from typing import Dict, List

from sklearn.ensemble import RandomForestClassifier

from sklearn.linear_model import LogisticRegression

from sklearn.model_selection import cross_val_score, StratifiedKFold

from sklearn.metrics import (roc_auc_score, roc_curve, confusion_matrix,

                             classification_report, accuracy_score)

 

 

class DefectClassifier:

    """随机森林分类未熔合缺陷"""

 

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

        self.random_state = random_state

        self.rf = RandomForestClassifier(

            n_estimators=150, max_depth=8,

            class_weight="balanced",

            random_state=random_state)

        self.lr = LogisticRegression(class_weight="balanced",

                                     random_state=random_state)

        self.feature_names_ = None

        self.selected_features_ = None

 

    def prepare_features(self, df: pd.DataFrame) -> np.ndarray:

        """选择特征列"""

        exclude = ["weld_id", "is_lack_of_fusion"]

        feature_cols = [c for c in df.columns if c not in exclude]

        self.feature_names_ = feature_cols

        return df[feature_cols].values

 

    def fit(self, X: np.ndarray, y: np.ndarray):

        self.rf.fit(X, y)

        self.lr.fit(X, y)

 

    def predict_proba_fusion(self, X: np.ndarray) -> np.ndarray:

        """未熔合概率"""

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

 

    def evaluate(self, X: np.ndarray, y: np.ndarray) -> Dict:

        """交叉验证评估"""

        skf = StratifiedKFold(n_splits=5, shuffle=True,

                              random_state=self.random_state)

        cv_auc = cross_val_score(self.rf, X, y, cv=skf, scoring="roc_auc")

        cv_f1 = cross_val_score(self.rf, X, y, cv=skf,

                                scoring="f1")

        y_pred = self.rf.predict(X)

        y_prob = self.predict_proba_fusion(X)

        fpr, tpr, _ = roc_curve(y, y_prob)

        return {

            "cv_auc_mean": round(cv_auc.mean(), 3),

            "cv_auc_std": round(cv_auc.std(), 3),

            "cv_f1_mean": round(cv_f1.mean(), 3),

            "train_accuracy": round(accuracy_score(y, y_pred), 3),

            "fpr": fpr,

            "tpr": tpr,

            "y_pred": y_pred,

            "y_prob": y_prob,

        }

 

    def feature_importance(self, top_n: int = 10) -> Dict:

        importances = self.rf.feature_importances_

        return dict(sorted(

            zip(self.feature_names_, importances),

            key=lambda x: x[1], reverse=True)[:top_n])

 

</details>

 

<details>

 

<summary></summary>

 

"""缺陷因果网络 (networkx)"""

 

import networkx as nx

import numpy as np

from typing import Dict

 

 

class DefectCausalNetwork:

    """构建工艺参数→波形特征→缺陷因果网络"""

 

    def __init__(self):

        self.G = nx.DiGraph()

 

    def build(self, feature_importance: Dict,

              defect_rate: float) -> nx.DiGraph:

        self.G.clear()

        # 缺陷节点

        self.G.add_node("未熔合", ntype="defect", weight=2.0)

        # 波形特征节点

        for feat, imp in list(feature_importance.items())[:6]:

            self.G.add_node(feat, ntype="wave_feature", weight=imp)

            self.G.add_edge(feat, "未熔合", weight=imp)

        # 工艺参数节点(连接波形特征)

        params = ["gap_mm", "interpass_temp_c", "wire_feed_m_min"]

        param_weights = {"gap_mm": 0.18, "interpass_temp_c": 0.10,

                         "wire_feed_m_min": 0.08}

        for p in params:

            self.G.add_node(p, ntype="process_param",

                          weight=param_weights.get(p, 0.05))

            # 连接到最相关的波形特征

            if "start_mean" in feature_importance:

                self.G.add_edge(p, "start_mean",

                               weight=param_weights.get(p, 0.05)*0.5)

        return self.G

 

</details>

 

<details>

 

<summary></summary>

 

"""可视化 (matplotlib + networkx)"""

 

import numpy as np

import pandas as pd

import matplotlib.pyplot as plt

from pathlib import Path

import networkx as nx

 

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

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

 

 

class WeldVisualizer:

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

        self.results_dir = Path(results_dir)

        self.results_dir.mkdir(exist_ok=True)

 

    def waveform_overlay(self, df: pd.DataFrame,

                         sample_welds: List[str] = None):

        """电流波形叠加缺陷标记"""

        if sample_welds is None:

            # 选2道合格+2道未熔合

            ok_ids = df[df["is_lack_of_fusion"]==0]["weld_id"].unique()[:2]

            bad_ids = df[df["is_lack_of_fusion"]==1]["weld_id"].unique()[:2]

            sample_welds = list(ok_ids) + list(bad_ids)

        fig, axes = plt.subplots(2, 2, figsize=(14, 10))

        axes = axes.flatten()

        for i, wid in enumerate(sample_welds[:4]):

            if i >= 4:

                break

            ax = axes[i]

            wdf = df[df["weld_id"]==wid].sort_values("timestamp_ms")

            is_bad = wdf["is_lack_of_fusion"].iloc[0]

            color = "#E74C3C" if is_bad else "#27AE60"

            ax.plot(wdf["timestamp_ms"]/1000, wdf["current_a"],

                   color=color, linewidth=1.2, alpha=0.8)

            ax.axvline(x=3.0, color="orange", linestyle="--", linewidth=1,

                      label="起弧段结束" if i==0 else "")

            ax.set_xlabel("时间 (s)")

            ax.set_ylabel("电流 (A)")

            ax.set_title(f"焊缝 {wid} ({'未熔合' if is_bad else '合格'})",

                        color=color, fontsize=11)

            ax.grid(alpha=0.3)

            if i == 0:

                ax.legend(fontsize=8)

        plt.suptitle("焊接电流波形(合格 vs 未熔合)", fontsize=14, fontweight="bold")

        plt.tight_layout()

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

        plt.close()

 

    def feature_boxplot(self, feat_df: pd.DataFrame):

        """特征分布箱线图"""

        fig, axes = plt.subplots(2, 3, figsize=(15, 10))

        axes = axes.flatten()

        features = ["start_mean", "start_current_drop", "gap_mm",

                    "steady_mean", "start_min", "interpass_temp_c"]

        for i, feat in enumerate(features):

            if i >= 6 or feat not in feat_df.columns:

                break

            ax = axes[i]

            ok_data = feat_df[feat_df["is_lack_of_fusion"]==0][feat].dropna()

            bad_data = feat_df[feat_df["is_lack_of_fusion"]==1][feat].dropna()

            bp = ax.boxplot([ok_data, bad_data], labels=["合格", "未熔合"],

                           patch_artist=True)

            bp["boxes"][0].set_facecolor("#27AE60")

            bp["boxes"][1].set_facecolor("#E74C3C")

            bp["boxes"][0].set_alpha(0.7)

            bp["boxes"][1].set_alpha(0.7)

            ax.set_title(feat, fontsize=10)

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

        plt.suptitle("关键特征分布(合格 vs 未熔合)", fontsize=13, fontweight="bold")

        plt.tight_layout()

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

        plt.close()

 

    def roc_curve(self, fpr, tpr, auc_score: float):

        """ROC曲线"""

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

        ax.plot(fpr, tpr, "b-", linewidth=2.5, label=f"ROC (AUC={auc_score:.3f})")

        ax.plot([0,1], [0,1], "k--", linewidth=1.5)

        ax.set_xlabel("假阳性率 (FPR)")

        ax.set_ylabel("真阳性率 (TPR)")

        ax.set_title("ROC 曲线 - 未熔合缺陷预判", fontsize=13, fontweight="bold")

        ax.legend(loc="lower right")

        ax.grid(alpha=0.3)

        plt.tight_layout()

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

        plt.close()

 

    def confusion_matrix(self, cm: np.ndarray):

        """混淆矩阵"""

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

        im = ax.imshow(cm, cmap="Blues", aspect="auto")

        ax.set_xticks([0, 1])

        ax.set_yticks([0, 1])

        ax.set_xticklabels(["合格", "未熔合"])

        ax.set_yticklabels(["合格", "未熔合"])

        ax.set_xlabel("预测")

        ax.set_ylabel("实际")

        for i in range(2):

            for j in range(2):

                ax.text(j, i, str(cm[i, j]), ha="center", va="center",

                       fontsize=16, fontweight="bold",

                       color="white" if cm[i, j] > cm.max()/2 else "black")

        plt.colorbar(im, ax=ax)

        ax.set_title("混淆矩阵", fontsize=13, fontweight="bold")

        plt.tight_layout()

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

        plt.close()

 

    def feature_importance(self, importance: Dict):

        """特征重要性"""

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

        names = list(importance.keys())

        vals = list(importance.values())

        colors = plt.cm.Reds(np.array(vals) / max(vals))

        ax.barh(range(len(names)), vals[::-1], color=colors[::-1],

                edgecolor="black", height=0.6)

        ax.set_yticks(range(len(names)))

        ax.set_yticklabels(names[::-1], fontsize=10)

        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 causal_network(self, G: nx.DiGraph):

        """因果网络图"""

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

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

        node_colors = []

        node_sizes = []

        for n in G.nodes():

            nt = G.nodes[n].get("ntype", "")

            if nt == "defect":

                node_colors.append("#E74C3C"); node_sizes.append(2500)

            elif nt == "wave_feature":

                node_colors.append("#3498DB"); node_sizes.append(1200)

            else:

                node_colors.append("#F39C12"); node_sizes.append(1000)

        nx.draw_networkx_nodes(G, pos, node_color=node_colors,

                              node_size=node_sizes, alpha=0.85, ax=ax)

        nx.draw_networkx_edges(G, pos, arrows=True, arrowsize=15,

                              edge_color="gray", alpha=0.5, ax=ax)

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

        ax.set_title("工艺参数→波形特征→未熔合 因果网络",

                    fontsize=14, fontweight="bold")

        ax.axis("off")

        plt.tight_layout()

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

        plt.close()

 

</details>

 

<details>

 

<summary></summary>

 

"""合成焊接数据"""

 

import numpy as np

import pandas as pd

from

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