周三下午，焊接机器人单元。

"这个月又撞了 9 次，"设备员大伟蹲在控制柜边翻报警本，"都是'轴5力矩超限+干涉报警'，但系统就记一条报警码，不告诉我在哪个点位撞的。维修靠回忆：'好像是取件位附近'、'可能是变位机那侧'。"

我插上 U 盘导出报警日志。

"这里面有啥？"我问。

"每条报警带时间戳、机器人编号、报警码、当前各轴角度、世界坐标 XYZ，"大伟说，"但就是堆在表里，没人按月份汇总，也没人把坐标反算成'工位区域'。排完会问'到底哪片区域最危险'，只能说'前面那块吧'。"

"最亏的是改布局，"大伟补一句，"上次产线挪了变位机，之后碰撞点跟着变，老经验全废。又得靠撞几次重新摸出来，每次撞都是工件报废+夹具磕伤。"

"我就想干一件事，"大伟说，"把一年报警日志灌进去，按月统计碰撞次数，把每条碰撞的世界坐标映射到工位区域网格，画成热力图，再用统计说清'哪块区域显著高发'，最后给个区域风险等级，写进点检表。"

"机器人碰撞不是'报警多就完了'，是'坐标落在哪片空间'，"我接话，"像给车间地面画格子，看哪格被踩出坑。用 pandas 按月聚合，numpy 做坐标网格映射，scipy 做卡方+泊松检验，sklearn 做区域风险分类，matplotlib 出热力图+月度趋势，networkx 建'区域→相邻区域→风险传导'图。"

"对，"大伟点头，"要能说清'3月碰撞 9 次，B区取件位占 6 次，占 67%，比基准区高 4.8 倍，p<0.001；风险等级红区；建议加软限位+区域降速'。以后改布局，重跑一遍就出新图。"

"OOP 封好，"我开工程，"日志加载器、月度聚合器、点位区域映射器、区域风险分类器、可视化器，合成 1 年 × 4 台机器人 × 报警坐标，下载就能跑。"

敲了行原型：

# 目标: 机器人碰撞报警 → 按月统计 + 点位区域映射 + 区域风险

# 方法: 月度聚合 + 网格分区 + 卡方/泊松 + RF风险分类 + 热力图

 

大伟凑近看："那以后看报告：1月5次、3月9次，B区取件位红成一片，相邻C区也被带成黄区，RF宏F1=0.97，点检表直接挂B区每日查。不用再靠撞几次摸位置了。"

"对，"我接话，"机器人安全不是'看报警总数'，是'看坐标落进哪格'。数字孪生里挂这个区域热力，就是维修班的'防撞地图'。"

 

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

 

场景设定：多台六轴工业机器人协同作业（焊接/搬运/上下料），控制器记录碰撞类报警（力矩超限、干涉、奇异点保护）并附带各轴角度与世界坐标。现场仅按报警码统计，不反算空间区域，导致"高频碰撞点不可见、布局变更后经验失效、点检无靶向"。

 

现场原话（叙事化）：

 

"不是机器人不稳，"大伟说，"是它撞完只喊一声码，不说撞哪儿。我们拿坐标反推过一次，发现 6 次都落在取件位左边 200mm 那个角，之前全靠腿走出来。"

"最亏的是改线，"大伟说，"变位机挪了 30cm，老防撞经验清零，又撞了 4 次才摸清新红区。每次都是夹具磕瘪、工件废件。"

核心矛盾："报警码堆表 + 人工回忆位置 + 经验防撞" 与 "按月统计 + 坐标网格分区 + 区域显著性检验 + 风险等级分类 + 相邻区域传导分析" 之间的断层。

 

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

 

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

工业机器人技术基础：六轴机器人坐标系、工作空间、干涉与碰撞保护、软限位 碰撞坐标→工位区域映射+风险分级

智能制造与数字孪生：机器人工作空间数字映射、安全区可视化 区域热力图挂数字孪生体

先进制造技术基础：设备可靠性、故障统计过程控制 按月碰撞次数 + 泊松过程建模

FMS与先进生产管理：多机协同安全布局、点检管理 多机器人区域风险联合分析

先进制造新模式：数据驱动的安全知识沉淀 区域风险模型复用至新布局

 

一句话总结：我们需要一个"机器人碰撞报警日志→月度聚合+坐标网格分区+卡方/泊松检验+sklearn区域风险分类+networkx相邻传导"程序，实现从"看报警总数"到"看空间红区+月度趋势"的闭环。

 

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

 

3.1 问题本质：给机器人地盘画格子

 

把机器人工作空间想成铺了地砖的房间：

 

* 世界坐标 XYZ = 脚踩的位置

* 网格分区 = 把地面划成 A/B/C/D 区（按 X-Y 投影分块，Z 分层可选）

* 碰撞次数 = 哪块砖上摔过跟头

* 月度统计 = 每个月数一遍各砖摔几次

* 红区 = 摔得显著多的砖

* 你的目标 = 不是"报警 9 次"，而是"B区那块砖，摔了 6 次，是基准区的 4.8 倍，且统计显著"

 

3.2 业务逻辑 → 代码映射

 

读取机器人碰撞报警日志

   │

   ▼ RobotCollisionLoader (pandas)

读取表:

  报警ID, 时间, 机器人编号, 报警码,

  X, Y, Z, 轴1~轴6角度, 是否碰撞类

   │

   ▼ MonthlyAggregator (pandas + numpy)

按月聚合:

  月维度碰撞次数, 各机器人次数, 报警码分布

  生成月度趋势序列

   │

   ▼ ZoneMapper (numpy)

点位区域映射:

  按 X∈[-W,W], Y∈[-W,W] 切网格(如 0.4m×0.4m)

  映射为区域编码 A1/B2...

  支持按 Z 分层(地面层/抬升层)

   │

   ▼ ZoneRiskAnalyzer (scipy)

统计检验:

  卡方检验: 区域×碰撞 独立性

  泊松检验: 各区域 λ 是否超基准

  比例 z: 红区 vs 基准区发生率

  基准区 = 最低密度区

   │

   ▼ ZoneRiskClassifier (sklearn)

风险分级:

  特征: 月均次数, λ, 占全空间比, 相邻区密度, Z分布

  标签: 0绿区 / 1黄区 / 2红区

  RF分类 + 特征重要性 + 5折交叉验证

   │

   ▼ CollisionVisualizer (matplotlib + networkx)

可视化:

  1. 月度碰撞次数折线+柱

  2. XY平面区域热力图

  3. 区域风险等级散点图

  4. 各机器人碰撞分布柱

  5. 特征重要性条图

  6. 区域邻接风险传导网络图

   │

   ▼ SyntheticCollisionLog (numpy)

合成数据:

  1年 × 4台机器人

  机制: B区取件位高发, 相邻C区次高, 其余低

  含轴5力矩超限为主报警码

 

3.3 为什么不能"只看报警总数"

 

视角 问题

报警码统计 知道撞了几次，不知道撞哪

人工回忆 布局一改全废

单看月度 看不出空间聚集

坐标网格化 每 0.4m 一格，红区肉眼可见

卡方+泊松 证明"真高发"不是随机

相邻传导 B红会带黄C，改布局可预判

风险分级 直接写点检表

 

3.4 分析前后对比

 

维度 传统方式 本程序

碰撞定位 "前面那块吧" B2区，坐标反算精确到格

月度统计 手数Excel pandas 自动按月+分机器人

高发判定 感觉 卡方 p<0.001 + λ倍数

布局变更 重撞几次摸 重跑脚本出新图

点检依据 全区域扫 红区每日查，绿区周查

 

四、OOP 代码实现

 

4.1 项目结构

 

robot_collision_zone/

├── robot_collision_zone/

│ ├── __init__.py

│ ├── collision_loader.py # 日志加载

│ ├── monthly_aggregator.py # 月度聚合

│ ├── zone_mapper.py # 坐标→网格区域(numpy)

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

│ ├── zone_risk_classifier.py # 风险分级(sklearn)

│ ├── collision_visualizer.py # 可视化

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

├── tests/

│ ├── __init__.py

│ └── test_collision.py

├── results/

│ ├── monthly_trend.png

│ ├── xy_heatmap.png

│ ├── zone_risk_scatter.png

│ ├── robot_dist_bar.png

│ ├── feature_importance.png

│ ├── zone_network.png

│ ├── zone_risk_detail.csv

│ └── collision_report.txt

└── run_collision.py

 

4.2 核心源码

 

<details>

 

<summary></summary>

 

"""机器人碰撞报警日志加载器。"""

 

import pandas as pd

from pathlib import Path

from typing import Set

 

 

class RobotCollisionLoader:

    """加载碰撞类报警日志。"""

 

    COLLISION_CODES: Set[str] = {

        "E0501", # 轴力矩超限

        "E0502", # 干涉报警

        "E0503", # 奇异点保护

        "E0510", # 软限位触发

    }

 

    def __init__(self, filepath: str = "robot_collision_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 = ["ts", "robot_id", "alarm_code", "x", "y", "z"]

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

        if miss:

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

        df["ts"] = pd.to_datetime(df["ts"], errors="coerce")

        df = df.dropna(subset=["ts", "x", "y", "z"]).reset_index(drop=True)

        df["is_collision"] = df["alarm_code"].isin(self.COLLISION_CODES).astype(int)

        df = df[df["is_collision"] == 1].copy()

        df["month"] = df["ts"].dt.to_period("M").astype(str)

        for c in ["x", "y", "z"]:

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

        return df

 

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

        s = f"碰撞记录数: {len(df)}\n"

        s += f"机器人: {sorted(df['robot_id'].unique())}\n"

        s += f"时间范围: {df['ts'].min()} ~ {df['ts'].max()}\n"

        s += f"月份数: {df['month'].nunique()}\n"

        s += f"报警码分布:\n{df['alarm_code'].value_counts().to_string()}"

        return s

 

</details>

 

<details>

 

<summary></summary>

 

"""按月聚合碰撞次数 (pandas)。"""

 

import pandas as pd

import numpy as np

from typing import Dict

 

 

class MonthlyAggregator:

    """月度碰撞统计 + 分机器人拆分。"""

 

    def __init__(self):

        pass

 

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

        g = (df.groupby("month")

               .agg(collision_cnt=("alarm_id", "count"),

                    robot_cnt=("robot_id", "nunique"))

               .reset_index())

        # 环比

        g["mom_delta"] = g["collision_cnt"].diff().fillna(0).astype(int)

        g["avg_per_robot"] = g["collision_cnt"] / g["robot_cnt"]

        return g

 

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

        return (df.groupby(["month", "robot_id"])

                  .size()

                  .reset_index(name="cnt"))

 

    def by_month_zone(self, df: pd.DataFrame, zone_col: str = "zone") -> pd.DataFrame:

        return (df.groupby(["month", zone_col])

                  .size()

                  .reset_index(name="cnt"))

 

</details>

 

<details>

 

<summary></summary>

 

"""世界坐标 → 网格区域映射 (numpy)。"""

 

import numpy as np

import pandas as pd

from typing import Tuple

 

 

class ZoneMapper:

    """按 X-Y 平面切网格, 支持 Z 分层。"""

 

    def __init__(self, half_span: float = 1.2, cell: float = 0.4,

                 z_layers: Tuple[float, float] = (0.0, 1.5)):

        """

        half_span: 工作空间半边长(m), 默认 ±1.2m

        cell: 网格边长(m)

        z_layers: Z分层边界

        """

        self.half_span = half_span

        self.cell = cell

        self.z_layers = z_layers

        self.n = int(2 * half_span // cell)

 

    def _idx(self, coord: np.ndarray) -> np.ndarray:

        # 映射到 [0, n)

        idx = np.floor((coord + self.half_span) / self.cell).astype(int)

        return np.clip(idx, 0, self.n - 1)

 

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

        d = df.copy()

        x = d["x"].values.astype(float)

        y = d["y"].values.astype(float)

        z = d["z"].values.astype(float)

        ix = self._idx(x)

        iy = self._idx(y)

        # 区域编码: 行=Y, 列=X, 如 A1

        rows = iy

        cols = ix

        letters = [chr(ord("A") + r) for r in rows]

        d["zone"] = [f"{letters[i]}{cols[i]+1}" for i in range(len(d))]

        d["grid_x"] = ix

        d["grid_y"] = iy

        # Z层

        d["z_layer"] = np.where(z < self.z_layers[1], "地面层", "抬升层")

        return d

 

    def zone_center(self, zone: str) -> Tuple[float, float]:

        """返回区域中心世界坐标"""

        row = ord(zone[0]) - ord("A")

        col = int(zone[1:]) - 1

        cx = -self.half_span + (col + 0.5) * self.cell

        cy = -self.half_span + (row + 0.5) * self.cell

        return cx, cy

 

</details>

 

<details>

 

<summary></summary>

 

"""区域风险统计检验 (scipy)。"""

 

import numpy as np

import pandas as pd

from typing import Dict

from scipy import stats

 

 

class ZoneRiskAnalyzer:

    """卡方 / 泊松 / 比例z 检验。"""

 

    @staticmethod

    def chi_square_zone(zone_df: pd.DataFrame) -> Dict:

        """区域 × 是否碰撞(此处用区域计数 vs 期望均匀)"""

        cnt = zone_df.groupby("zone").size()

        obs = cnt.values.astype(float)

        exp = np.full_like(obs, obs.mean())

        if len(obs) < 2:

            return {"valid": False}

        chi2, p = stats.chisquare(obs, exp)

        return {"valid": True, "chi2": float(chi2), "p_value": float(p),

               "n_zones": len(obs)}

 

    @staticmethod

    def poisson_lambda(zone_df: pd.DataFrame) -> pd.DataFrame:

        """各区域泊松 λ(月均碰撞次数)"""

        tmp = zone_df.copy()

        tmp["month"] = tmp["ts"].dt.to_period("M")

        g = (tmp.groupby("zone")

                 .agg(total=("alarm_id", "count"),

                      months=("month", "nunique"))

                 .reset_index())

        g["lambda_per_month"] = g["total"] / g["months"]

        return g

 

    @staticmethod

    def vs_baseline(zone_cnt: pd.Series, zone: str) -> Dict:

        """某区域 vs 最低密度基准区 比例z检验"""

        base_zone = zone_cnt.idxmin()

        x1 = int(zone_cnt[zone]); n1 = int(zone_cnt.sum())

        x2 = int(zone_cnt[base_zone]); n2 = int(zone_cnt.sum())

        # 用占比近似

        p1, p2 = x1/n1, x2/n2

        p_pool = (x1 + x2) / (n1 + n2)

        se = np.sqrt(p_pool*(1-p_pool)*(1/n1 + 1/n2))

        if se == 0:

            return {"z": 0.0, "p_value": 1.0, "ratio": 1.0, "base_zone": base_zone}

        z = (p1 - p2) / se

        p = 2 * (1 - stats.norm.cdf(abs(z)))

        return {"z": float(z), "p_value": float(p),

               "ratio": float(x1/(x2+1e-9)), "base_zone": base_zone}

 

    @staticmethod

    def adjacent_matrix(zones: list, n_row: int) -> np.ndarray:

        """生成网格邻接矩阵(上下左右)"""

        idx = {z: (ord(z[0])-ord("A"), int(z[1:])-1) for z in zones}

        m = np.zeros((len(zones), len(zones)), dtype=int)

        for i, zi in enumerate(zones):

            r, c = idx[zi]

            for j, zj in enumerate(zones):

                rr, cc = idx[zj]

                if abs(r-rr) + abs(c-cc) == 1:

                    m[i, j] = 1

        return m

 

</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.model_selection import cross_val_score

from sklearn.inspection import permutation_importance

 

 

class ZoneRiskClassifier:

    """按区域特征分绿/黄/红。"""

 

    LABEL = {0: "绿区", 1: "黄区", 2: "红区"}

 

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

        self.random_state = random_state

        self.model_ = None

 

    def build_features(self, zone_df: pd.DataFrame,

                      adj_matrix: np.ndarray,

                      zones: List[str]) -> pd.DataFrame:

        g = (zone_df.groupby("zone")

                    .agg(monthly_avg=("alarm_id", "count"),

                         x=("x", "mean"),

                         y=("y", "mean"))

                    .reset_index())

        # 月均需再除月份数

        months = zone_df["month"].nunique()

        g["monthly_avg"] = g["monthly_avg"] / months

        # 邻接区密度和

        cnt = zone_df.groupby("zone").size()

        adj_density = []

        for i, z in enumerate(zones):

            nb = np.where(adj_matrix[i] == 1)[0]

            s = sum(cnt.get(zones[j], 0) for j in nb)

            adj_density.append(s)

        g["adj_density"] = adj_density

        g["ratio_to_base"] = g["monthly_avg"] / (g["monthly_avg"].min() + 1e-9)

        # 打标

        def lab(r):

            if r["monthly_avg"] >= 2.0 or r["ratio_to_base"] >= 3:

                return 2

            elif r["monthly_avg"] >= 0.8 or r["ratio_to_base"] >= 1.8:

                return 1

            return 0

        g["label"] = g.apply(lab, axis=1)

        return g

 

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

        self.model_ = RandomForestClassifier(

            n_estimators=200, max_depth=5, min_samples_leaf=2,

            class_weight="balanced", random_state=self.random_state, n_jobs=-1)

        self.model_.fit(X, y)

        return self.model_

 

    def feature_importance(self, feat_cols: List[str]) -> Dict:

        imp = dict(zip(feat_cols, self.model_.feature_importances_))

        return dict(sorted(imp.items(), key=lambda x: x[1], reverse=True))

 

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

        sc = cross_val_score(self.model_, X, y, cv=5, scoring="f1_macro")

        return {"f1_macro_mean": float(sc.mean()), "f1_macro_std": float(sc.std())}

 

</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", "WenQuanYi Micro Hei", "DejaVu Sans"]

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

 

RISK_COLOR = {0: "#27AE60", 1: "#F39C12", 2: "#E74C3C"}

 

 

class CollisionVisualizer:

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

        self.results_dir = Path(results_dir)

        self.results_dir.mkdir(exist_ok=True)

 

    def monthly_trend(self, mdf: pd.DataFrame):

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

        ax1.bar(mdf["month"], mdf["collision_cnt"], color="#3498DB",

               alpha=0.8, label="月度碰撞次数")

        ax1.set_ylabel("碰撞次数", fontsize=12)

        ax2 = ax1.twinx()

        ax2.plot(mdf["month"], mdf["avg_per_robot"], "o-", color="#E74C3C",

                label="单机器人均值")

        ax2.set_ylabel("单机器人均值", fontsize=12)

        ax1.set_title("月度碰撞次数趋势", fontsize=13, fontweight="bold")

        ax1.tick_params(axis="x", rotation=45)

        ax1.legend(loc="upper left"); ax2.legend(loc="upper right")

        ax1.grid(alpha=0.3)

        plt.tight_layout()

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

                   dpi=150, bbox_inches="tight")

        plt.close()

 

    def xy_heatmap(self, zdf: pd.DataFrame, zm: "ZoneMapper"):

        """XY平面碰撞热力图"""

        grid = np.zeros((zm.n, zm.n))

        for _, r in zdf.iterrows():

            grid[r["grid_y"], r["grid_x"]] += 1

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

        im = ax.imshow(grid, origin="lower", cmap="YlOrRd")

        ax.set_xticks(range(zm.n))

        ax.set_xticklabels([f"{i+1}" for i in range(zm.n)])

        ax.set_yticks(range(zm.n))

        ax.set_yticklabels([chr(ord("A")+i) for i in range(zm.n)])

        ax.set_xlabel("X网格列", fontsize=12)

        ax.set_ylabel("Y网格行", fontsize=12)

        for i in range(zm.n):

            for j in range(zm.n):

                if grid[i, j] > 0:

                    ax.text(j, i, int(grid[i, j]), ha="center", va="center",

                           fontsize=9, color="black")

        ax.set_title("XY平面碰撞密度热力图", fontsize=13, fontweight="bold")

        plt.colorbar(im, ax=ax, fraction=0.046, label="碰撞次数")

        plt.tight_layout()

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

                   dpi=150, bbox_inches="tight")

        plt.close()

 

    def zone_risk_scatter(self, feat: pd.DataFrame):

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

        colors = [RISK_COLOR[l] for l in feat["label"]]

        ax.scatter(feat["x"], feat["y"], s=feat["monthly_avg"]*120+30,

                  c=colors, alpha=0.8, edgecolors="k")

        for _, r in feat.iterrows():

            ax.text(r["x"], r["y"], r["zone"], fontsize=9, ha="center",

                   va="center", color="white", fontweight="bold")

        ax.set_xlabel("世界坐标 X (m)", fontsize=12)

        ax.set_ylabel("世界坐标 Y (m)", fontsize=12)

        ax.set_title("区域风险等级(气泡=月均次数)",

                    fontsize=13, fontweight="bold")

        ax.grid(alpha=0.3)

        plt.tight_layout()

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

                   dpi=150, bbox_inches="tight")

        plt.close()

 

    def robot_dist_bar(self, df: pd.DataFrame):

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

        g = df.groupby("robot_id").size()

        ax.bar(g.index, g.values, color="#2980B9", edgecolor="black")

        for i, v in enumerate(g.values):

            ax.text(i, v, str(v), ha="center", va="bottom", fontsize=11)

        ax.set_ylabel("碰撞次数", fontsize=12)

        ax.set_title("各机器人碰撞分布", fontsize=13, fontweight="bold")

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

        plt.tight_layout()

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

                   dpi=150, bbox_inches="tight")

        plt.close()

 

    def feature_importance_bar(self, imp: Dict):

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

        keys = list(imp.keys())[:6]

        vals = [imp[k] for k in keys]

        ax.barh(keys[::-1], vals[::-1], color="#8E44AD", alpha=0.85)

        for i, v in enumerate(vals[::-1]):

            ax.text(v, i, f"{v:.3f}", va="center", fontsize=9)

        ax.set_xlabel("特征重要性", fontsize=12)

        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 zone_network(self, feat: pd.DataFrame, adj: np.ndarray, zones: list):

        """区域邻接风险传导网络"""

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

        G = nx.Graph()

        for z in zones:

            lab = feat[feat["zone"] == z]["label"].values[0]

            G.add_node(z, risk=int(lab))

        for i in range(len(zones)):

            for j in range(i+1, len(zones)):

                if adj[i, j] == 1:

                    G.add_edge(zones[i], zones[j])

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

        nc = [RISK_COLOR[G.nodes[n]["risk"]] for n in G.nodes()]

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

                              alpha=0.9, ax=ax)

        nx.draw_networkx_edges(G, pos, edge_color="#7F8C8D",

                              width=1.5, alpha=0.6, ax=ax)

        nx.draw_networkx_labels(G, pos, font_size=10, ax=ax,

                              font_color="white", font_weight="bold")

        ax.set_title("区域邻接风险传导网络", fontsize=14, fontweight="bold")

        ax.axis("off")

        plt.tight_layout()

        plt.savefig(self.results_dir/"zone_network.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 SyntheticCollisionLog:

    """

    生成1年 × 4台机器人碰撞日志

    机制:

      B2区(取件位左侧)高发, 相邻A2/C2次高

      报警码以 E0501(轴5力矩超限)为主

      月度有波动, 3月最高

    """

 

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

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

 

    def generate(self, out_path: str = "robot_collision_log.csv",

                year: int = 2025, n_robots: int = 4) -> pd.DataFrame:

        records = []

        robots = [f"R0{i}" for i in range(1, n_robots+1)]

        alarm_pool = ["E0501", "E0502", "E0503", "E0510"]

        alarm_w = [0.62, 0.22, 0.1, 0.06]

        aid = 0

        # 定义热点: B2区中心约 x=-0.4~0, y=0~0.4

        for m in range(1, 13):

            # 月度基础次数

            base = {3: 9, 6: 7, 9: 6}.get(m, self.rng.randint(3, 6))

            for _ in range(base):

                aid += 1

                # 70%落在B2, 20%相邻, 10%散落

                u = self.rng.uniform(0, 1)

                if u < 0.7:

                    x = self.rng.normal(-0.2, 0.08)

                    y = self.rng.normal(0.2, 0.08)

                elif u < 0.9:

                    x = self.rng.normal(-0.6, 0.1)

                    y = self.rng.normal(0.2, 0.1)

                else:

                    x = self.rng.normal(0, 0.5)

                    y = self

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