From 7961c4a86ad316f1b5290f13413bd9432a853e5c Mon Sep 17 00:00:00 2001 From: aliangabc Date: Mon, 22 Dec 2025 12:15:56 +0800 Subject: [PATCH 1/8] aaa --- .../speed_main.py | 293 ++++++++++++------ 1 file changed, 195 insertions(+), 98 deletions(-) diff --git a/src/unmanned_vehicle_safety_and_operation/speed_main.py b/src/unmanned_vehicle_safety_and_operation/speed_main.py index ab7726cc9e..cc76926ecf 100644 --- a/src/unmanned_vehicle_safety_and_operation/speed_main.py +++ b/src/unmanned_vehicle_safety_and_operation/speed_main.py @@ -1,17 +1,20 @@ -import numpy as np -import matplotlib.pyplot as plt -import random -from typing import Tuple, Dict, List -from sklearn.linear_model import LinearRegression -from sklearn.ensemble import IsolationForest, RandomForestClassifier -from sklearn.model_selection import train_test_split -from sklearn.metrics import mean_absolute_error, accuracy_score, classification_report -import warnings - -warnings.filterwarnings("ignore") # 忽略无关警告 - - -# ===================== 基础数据生成 ===================== +# 导入必要的库 +import numpy as np # 数值计算基础库 +import matplotlib.pyplot as plt # 数据可视化库 +import random # 随机数生成库 +from typing import Tuple, Dict, List # 类型注解库 +# 机器学习相关库 +from sklearn.linear_model import LinearRegression # 线性回归模型 +from sklearn.ensemble import IsolationForest, RandomForestClassifier # 孤立森林（异常检测）、随机森林（分类） +from sklearn.model_selection import train_test_split # 数据集划分 +from sklearn.metrics import mean_absolute_error, accuracy_score, classification_report # 评估指标 +import warnings # 警告处理库 + +# 全局设置：忽略无关警告（提升代码运行整洁性） +warnings.filterwarnings("ignore") + + +# ===================== 基础数据生成模块 ===================== def generate_velocity_data( test_duration: int = 60, sample_freq: int = 1, @@ -19,151 +22,218 @@ def generate_velocity_data( noise_level: float = 0.5, add_abnormal: bool = True # 是否加入异常值（模拟故障） ) -> Tuple[np.ndarray, np.ndarray]: - """生成模拟无人车速度数据（含可选异常值）""" + """ + 生成模拟无人车速度时序数据，模拟真实驾驶的加速-匀速-减速过程，并可添加异常值 + + 参数说明： + -------- + test_duration : int + 测试总时长（秒），默认60秒 + sample_freq : int + 采样频率（次/秒），默认1次/秒 + max_velocity : float + 最大行驶速度（km/h），默认30km/h + noise_level : float + 速度噪声幅度（模拟传感器误差），默认±0.5km/h + add_abnormal : bool + 是否添加异常值（模拟传感器故障/急刹等异常），默认True + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray] + time_steps: 时间戳数组（秒） + velocity: 对应时间的速度数组（km/h） + """ + # 生成时间戳序列：从0到test_duration，步长为1/sample_freq time_steps = np.arange(0, test_duration, 1 / sample_freq) - num_samples = len(time_steps) - velocity = np.zeros(num_samples) + num_samples = len(time_steps) # 总采样点数 + velocity = np.zeros(num_samples) # 初始化速度数组 - # 模拟加速→匀速→减速 + # 模拟典型驾驶过程：0-10秒加速 | 10-40秒匀速 | 40-60秒减速 for i, t in enumerate(time_steps): if t < 10: + # 加速阶段：线性增长到最大速度 v = (max_velocity / 10) * t elif 10 <= t < 40: + # 匀速阶段：保持最大速度 v = max_velocity else: + # 减速阶段：线性减速至0 v = max_velocity - (max_velocity / 20) * (t - 40) - # 加入噪声 + # 添加随机噪声（模拟传感器测量误差） v += random.uniform(-noise_level, noise_level) + # 确保速度非负（物理意义：速度不能为负） velocity[i] = max(v, 0.0) - # 随机加入2-3个异常值（模拟传感器故障/急刹） + # 随机添加2-3个异常值（模拟传感器故障/急刹/数据跳变） if add_abnormal: + # 随机选择2-3个异常点位置 abnormal_idx = random.sample(range(num_samples), random.randint(2, 3)) for idx in abnormal_idx: - velocity[idx] = velocity[idx] * random.uniform(2, 3) # 异常飙升 + # 异常值类型1：速度飙升（2-3倍当前值） + velocity[idx] = velocity[idx] * random.uniform(2, 3) + # 50%概率触发异常值类型2：速度归零（模拟急刹/传感器掉线） if random.random() > 0.5: - velocity[idx] = 0 # 异常归零 + velocity[idx] = 0 return time_steps, velocity -# ===================== 机器学习：速度预测 ===================== +# ===================== 机器学习：速度预测模块 ===================== def velocity_prediction( time_steps: np.ndarray, velocity: np.ndarray, predict_ratio: float = 0.2 # 预测未来20%的数据 ) -> Tuple[np.ndarray, np.ndarray, float]: """ - 基于线性回归预测未来速度 - :param time_steps: 时间戳数组 - :param velocity: 速度数组 - :param predict_ratio: 预测未来数据占比 - :return: 预测时间戳、预测速度、MAE误差 + 基于线性回归模型对无人车未来速度进行预测 + + 参数说明： + -------- + time_steps : np.ndarray + 历史时间戳数组 + velocity : np.ndarray + 历史速度数组 + predict_ratio : float + 预测未来数据占总数据的比例，默认0.2（预测未来20%） + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray, float] + predict_time: 预测时间段的时间戳数组 + predict_vel: 预测的速度值数组 + mae: 平均绝对误差（衡量预测精度） """ - # 数据预处理：划分训练/预测时间范围 + # 数据预处理：划分训练集和预测集 total_len = len(time_steps) - train_len = int(total_len * (1 - predict_ratio)) - train_time = time_steps[:train_len].reshape(-1, 1) # 回归要求特征为2D - train_vel = velocity[:train_len] - predict_time = time_steps[train_len:].reshape(-1, 1) + train_len = int(total_len * (1 - predict_ratio)) # 训练集长度（80%数据） + # 训练集特征：时间戳（需reshape为2D数组，满足sklearn输入要求） + train_time = time_steps[:train_len].reshape(-1, 1) + train_vel = velocity[:train_len] # 训练集标签：对应时间的速度 + predict_time = time_steps[train_len:].reshape(-1, 1) # 预测时间段 - # 训练线性回归模型 + # 初始化并训练线性回归模型 model = LinearRegression() - model.fit(train_time, train_vel) + model.fit(train_time, train_vel) # 拟合时间-速度关系 # 预测未来速度 predict_vel = model.predict(predict_time) - # 计算预测误差（仅当有真实值时） + # 计算预测误差（仅当有真实值可对比时） if len(predict_vel) == len(velocity[train_len:]): mae = mean_absolute_error(velocity[train_len:], predict_vel) else: - mae = 0.0 + mae = 0.0 # 无真实值时误差为0 + # 展平时间数组（便于后续可视化） return predict_time.flatten(), predict_vel, mae -# ===================== 机器学习：异常检测 ===================== +# ===================== 机器学习：异常检测模块 ===================== def detect_abnormal_velocity( velocity: np.ndarray, contamination: float = 0.05 # 异常值占比（默认5%） ) -> np.ndarray: """ - 基于孤立森林检测速度异常值 - :param velocity: 速度数组 - :param contamination: 异常值比例 - :return: 异常标记数组（1=正常，-1=异常） + 基于孤立森林（Isolation Forest）算法检测速度序列中的异常值 + 孤立森林原理：通过随机划分特征空间，孤立异常点（异常点更容易被孤立） + + 参数说明： + -------- + velocity : np.ndarray + 速度序列数组 + contamination : float + 预设异常值比例（0-1），默认0.05（5%） + + 返回值： + -------- + np.ndarray + abnormal_label: 异常标记数组（1=正常样本，-1=异常样本） """ - # 数据预处理：重塑为2D数组 + # 数据预处理：重塑为2D数组（sklearn模型输入要求） vel_2d = velocity.reshape(-1, 1) - # 训练孤立森林模型 + # 初始化孤立森林模型 model = IsolationForest( - n_estimators=100, - contamination=contamination, - random_state=42 + n_estimators=100, # 决策树数量，默认100 + contamination=contamination, # 异常值比例 + random_state=42 # 随机种子（保证结果可复现） ) + # 训练模型并预测异常值 abnormal_label = model.fit_predict(vel_2d) # 1=正常，-1=异常 return abnormal_label -# ===================== 机器学习：驾驶模式分类 ===================== +# ===================== 机器学习：驾驶模式分类模块 ===================== def classify_driving_mode( time_steps: np.ndarray, velocity: np.ndarray ) -> Tuple[np.ndarray, float]: """ - 基于随机森林分类驾驶模式（0=减速，1=匀速，2=加速） - :param time_steps: 时间戳数组 - :param velocity: 速度数组 - :return: 模式标签数组、分类准确率 + 基于随机森林分类器识别驾驶模式（加速/匀速/减速） + 核心特征：加速度（速度变化率）+ 当前速度 + + 参数说明： + -------- + time_steps : np.ndarray + 时间戳数组 + velocity : np.ndarray + 速度数组 + + 返回值： + -------- + Tuple[np.ndarray, float] + full_pred: 全量数据的驾驶模式标签（0=减速，1=匀速，2=加速） + accuracy: 分类准确率（测试集） """ - # 特征工程：计算加速度（速度差分/时间差分）作为核心特征 - vel_diff = np.diff(velocity) - time_diff = np.diff(time_steps) - acceleration = vel_diff / time_diff + # 特征工程：计算加速度（核心特征） + vel_diff = np.diff(velocity) # 速度差分（后-前） + time_diff = np.diff(time_steps) # 时间差分 + acceleration = vel_diff / time_diff # 加速度 = 速度变化/时间变化 - # 构建标签：根据加速度分类 - # 加速度>0.5 → 加速(2)；-0.5<加速度<0.5 → 匀速(1)；加速度<-0.5 → 减速(0) + # 构建标签：根据加速度阈值分类驾驶模式 labels = [] for acc in acceleration: if acc > 0.5: - labels.append(2) + labels.append(2) # 加速模式：加速度>0.5 km/h/s elif acc > -0.5: - labels.append(1) + labels.append(1) # 匀速模式：-0.5<加速度<0.5 km/h/s else: - labels.append(0) + labels.append(0) # 减速模式：加速度<-0.5 km/h/s labels = np.array(labels) - # 构建特征集：加速度+当前速度（需对齐长度） + # 构建特征集：加速度 + 当前速度（需对齐长度，差分后长度减1） features = np.column_stack([ acceleration, - velocity[:-1] # 差分后长度减1 + velocity[:-1] # 速度数组去掉最后一个元素，与加速度长度匹配 ]) - # 划分训练/测试集 + # 划分训练集（70%）和测试集（30%） X_train, X_test, y_train, y_test = train_test_split( features, labels, test_size=0.3, random_state=42 ) - # 训练随机森林分类器 - model = RandomForestClassifier(n_estimators=100, random_state=42) + # 初始化并训练随机森林分类器 + model = RandomForestClassifier( + n_estimators=100, # 决策树数量 + random_state=42 # 随机种子 + ) model.fit(X_train, y_train) - # 预测并计算准确率 + # 预测测试集并计算准确率 y_pred = model.predict(X_test) accuracy = accuracy_score(y_test, y_pred) - # 预测全量数据的模式标签（补全最后一个点的标签） + # 预测全量数据的驾驶模式（补全最后一个点的标签） full_pred = model.predict(features) - full_pred = np.append(full_pred, full_pred[-1]) # 补全长度 + full_pred = np.append(full_pred, full_pred[-1]) # 补全长度，保持与原始数据一致 return full_pred, accuracy -# ===================== 可视化整合 ===================== +# ===================== 可视化整合模块 ===================== def plot_ml_results( time_steps: np.ndarray, velocity: np.ndarray, @@ -172,59 +242,84 @@ def plot_ml_results( abnormal_label: np.ndarray, mode_labels: np.ndarray ): - """可视化机器学习分析结果""" + """ + 可视化所有机器学习分析结果：速度预测+异常检测+驾驶模式分类 + + 参数说明： + -------- + time_steps : np.ndarray + 原始时间戳数组 + velocity : np.ndarray + 原始速度数组 + predict_time : np.ndarray + 预测时间段的时间戳 + predict_vel : np.ndarray + 预测的速度值 + abnormal_label : np.ndarray + 异常检测标记数组 + mode_labels : np.ndarray + 驾驶模式分类标签数组 + """ + # 设置中文字体（避免中文乱码） plt.rcParams["font.sans-serif"] = ["SimHei"] plt.rcParams["axes.unicode_minus"] = False - # 创建2行1列的子图 + # 创建2行1列的子图布局，设置画布大小 fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) - # -------- 子图1：速度预测+异常检测 -------- - # 绘制原始速度 + # -------- 子图1：速度预测 + 异常检测可视化 -------- + # 绘制原始速度曲线 ax1.plot(time_steps, velocity, color="#2E86AB", linewidth=2, label="原始速度") - # 绘制预测速度 + # 绘制预测速度曲线（虚线） ax1.plot(predict_time, predict_vel, color="#F18F01", linestyle="--", linewidth=2, label="预测速度") - # 标注异常点 + # 标注异常点（红色高亮） abnormal_idx = np.where(abnormal_label == -1)[0] ax1.scatter(time_steps[abnormal_idx], velocity[abnormal_idx], - color="#E63946", s=100, label="异常速度点", zorder=5) + color="#E63946", s=100, label="异常速度点", zorder=5) # zorder确保异常点在顶层 + # 子图1样式设置 ax1.set_xlabel("时间 (秒)", fontsize=12) ax1.set_ylabel("速度 (km/h)", fontsize=12) ax1.set_title("无人车速度预测 + 异常检测", fontsize=14, fontweight="bold") ax1.legend(loc="upper right") - ax1.grid(True, alpha=0.3) + ax1.grid(True, alpha=0.3) # 添加网格（透明度0.3） - # -------- 子图2：驾驶模式分类 -------- - # 定义模式颜色和标签 - mode_colors = {0: "#A23B72", 1: "#2E86AB", 2: "#F18F01"} + # -------- 子图2：驾驶模式分类可视化 -------- + # 定义模式颜色和标签映射 + mode_colors = {0: "#A23B72", 1: "#2E86AB", 2: "#F18F01"} # 减速-紫 | 匀速-蓝 | 加速-橙 mode_names = {0: "减速", 1: "匀速", 2: "加速"} - # 绘制分类结果 + # 按驾驶模式绘制散点图 for mode in [0, 1, 2]: mode_idx = np.where(mode_labels == mode)[0] ax2.scatter(time_steps[mode_idx], velocity[mode_idx], color=mode_colors[mode], label=mode_names[mode], s=50, alpha=0.7) + # 子图2样式设置 ax2.set_xlabel("时间 (秒)", fontsize=12) ax2.set_ylabel("速度 (km/h)", fontsize=12) ax2.set_title("无人车驾驶模式分类", fontsize=14, fontweight="bold") ax2.legend(loc="upper right") ax2.grid(True, alpha=0.3) + # 调整子图间距，避免重叠 plt.tight_layout() + # 显示图形 plt.show() -# ===================== 主函数 ===================== +# ===================== 主函数（程序入口） ===================== def main(): - # 1. 配置参数 - TEST_DURATION = 60 - SAMPLE_FREQ = 2 # 提高采样频率（每秒2次） - MAX_VELOCITY = 30 - NOISE_LEVEL = 0.5 + """ + 程序主入口：整合所有模块，执行数据生成→模型训练→结果分析→可视化 + """ + # 1. 配置核心参数 + TEST_DURATION = 60 # 测试时长60秒 + SAMPLE_FREQ = 2 # 采样频率2次/秒（提高数据密度） + MAX_VELOCITY = 30 # 最大速度30km/h + NOISE_LEVEL = 0.5 # 速度噪声±0.5km/h - # 2. 生成测试数据 + # 2. 生成模拟无人车速度数据 print("===== 生成无人车速度测试数据 =====") time_steps, velocity = generate_velocity_data( test_duration=TEST_DURATION, @@ -234,33 +329,35 @@ def main(): add_abnormal=True ) - # 3. 速度预测 + # 3. 执行速度预测（线性回归） print("\n===== 速度预测（线性回归） =====") predict_time, predict_vel, mae = velocity_prediction(time_steps, velocity) print(f"预测未来{len(predict_time)}个时间点的速度") - print(f"预测平均绝对误差（MAE）：{mae:.2f} km/h") + print(f"预测平均绝对误差（MAE）：{mae:.2f} km/h") # 保留2位小数 - # 4. 异常检测 + # 4. 执行异常检测（孤立森林） print("\n===== 异常检测（孤立森林） =====") abnormal_label = detect_abnormal_velocity(velocity) - abnormal_num = len(np.where(abnormal_label == -1)[0]) + abnormal_num = len(np.where(abnormal_label == -1)[0]) # 统计异常点数量 print(f"检测到异常速度点数量：{abnormal_num} 个") - abnormal_time = time_steps[abnormal_label == -1] - print(f"异常点时间戳：{abnormal_time.round(2)}") + abnormal_time = time_steps[abnormal_label == -1] # 异常点时间戳 + print(f"异常点时间戳：{abnormal_time.round(2)}") # 保留2位小数 - # 5. 驾驶模式分类 + # 5. 执行驾驶模式分类（随机森林） print("\n===== 驾驶模式分类（随机森林） =====") mode_labels, accuracy = classify_driving_mode(time_steps, velocity) - print(f"分类准确率：{accuracy:.2%}") + print(f"分类准确率：{accuracy:.2%}") # 百分比格式显示准确率 + # 统计各模式数量 mode_count = {0: 0, 1: 0, 2: 0} for label in mode_labels: mode_count[label] += 1 print(f"减速模式次数：{mode_count[0]} | 匀速模式次数：{mode_count[1]} | 加速模式次数：{mode_count[2]}") - # 6. 可视化结果 + # 6. 可视化所有分析结果 print("\n===== 可视化分析结果 =====") plot_ml_results(time_steps, velocity, predict_time, predict_vel, abnormal_label, mode_labels) +# 程序执行入口 if __name__ == "__main__": main() \ No newline at end of file From 9613b87f8158d497c0bf1c764c42deeea24c8db6 Mon Sep 17 00:00:00 2001 From: aliangabc Date: Mon, 22 Dec 2025 19:20:44 +0800 Subject: [PATCH 2/8] aaa --- src/unmanned_vehicle_safety_and_operation/speed_main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/unmanned_vehicle_safety_and_operation/speed_main.py b/src/unmanned_vehicle_safety_and_operation/speed_main.py index cc76926ecf..62caace396 100644 --- a/src/unmanned_vehicle_safety_and_operation/speed_main.py +++ b/src/unmanned_vehicle_safety_and_operation/speed_main.py @@ -268,7 +268,7 @@ def plot_ml_results( fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) # -------- 子图1：速度预测 + 异常检测可视化 -------- - # 绘制原始速度曲线 + # 绘制原始速度曲线11 ax1.plot(time_steps, velocity, color="#2E86AB", linewidth=2, label="原始速度") # 绘制预测速度曲线（虚线） ax1.plot(predict_time, predict_vel, color="#F18F01", linestyle="--", linewidth=2, label="预测速度") From 2c8fbbc2cf4754864ebc06d8ade36cd4bfbdba15 Mon Sep 17 00:00:00 2001 From: aliangabc Date: Mon, 22 Dec 2025 19:56:46 +0800 Subject: [PATCH 3/8] aaa --- src/unmanned_vehicle_safety_and_operation/speed_main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/unmanned_vehicle_safety_and_operation/speed_main.py b/src/unmanned_vehicle_safety_and_operation/speed_main.py index 62caace396..b3df06e46c 100644 --- a/src/unmanned_vehicle_safety_and_operation/speed_main.py +++ b/src/unmanned_vehicle_safety_and_operation/speed_main.py @@ -284,7 +284,7 @@ def plot_ml_results( ax1.legend(loc="upper right") ax1.grid(True, alpha=0.3) # 添加网格（透明度0.3） - # -------- 子图2：驾驶模式分类可视化 -------- + # -------- 子图2：驾驶模式分类可视化 1-------- # 定义模式颜色和标签映射 mode_colors = {0: "#A23B72", 1: "#2E86AB", 2: "#F18F01"} # 减速-紫 | 匀速-蓝 | 加速-橙 mode_names = {0: "减速", 1: "匀速", 2: "加速"} From cfd5178b7e5d515976e12569514fd59670297396 Mon Sep 17 00:00:00 2001 From: aliangabc Date: Tue, 23 Dec 2025 00:25:14 +0800 Subject: [PATCH 4/8] aaa --- .../temperature_alarm.py | 523 ++++++++++++++---- 1 file changed, 408 insertions(+), 115 deletions(-) diff --git a/src/unmanned_vehicle_safety_and_operation/temperature_alarm.py b/src/unmanned_vehicle_safety_and_operation/temperature_alarm.py index 0a79b3c955..8b72c249a1 100644 --- a/src/unmanned_vehicle_safety_and_operation/temperature_alarm.py +++ b/src/unmanned_vehicle_safety_and_operation/temperature_alarm.py @@ -4,79 +4,141 @@ from collections import defaultdict import matplotlib.pyplot as plt -# 解决PyCharm中文显示问题 -plt.rcParams['font.sans-serif'] = ['SimHei'] # 黑体 -plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题 +# 解决PyCharm中文显示问题（全局字体配置） +plt.rcParams['font.sans-serif'] = ['SimHei'] # 设置默认字体为黑体 +plt.rcParams['axes.unicode_minus'] = False # 解决负号显示为方块的问题 # ===================== 1. 数据生成模块 ===================== def generate_temperature_data(n_samples=10000): - """生成模拟的无人车温度数据集""" + """ + 生成模拟的无人车关键部件温度数据集（含正常/异常样本） + 模拟场景：无人车电机、电池、控制器温度随运行时长/车速变化，同时注入不同等级的异常 + + 参数说明： + -------- + n_samples : int + 生成样本总数，默认10000条 + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray] + features: 特征数组 (n_samples, 5)，列顺序：电机温度、电池温度、控制器温度、运行时长、车速 + labels: 标签数组 (n_samples,)，取值：0=正常，1=轻度异常，2=中度异常，3=重度异常 + """ + # 设置随机种子（保证结果可复现） np.random.seed(42) - # 基础特征 - motor_temp = np.random.normal(60, 10, n_samples) - battery_temp = np.random.normal(45, 8, n_samples) - controller_temp = np.random.normal(50, 9, n_samples) - runtime = np.random.uniform(0, 10, n_samples) - speed = np.random.normal(40, 15, n_samples) - - # 构造异常数据 + + # ========== 生成基础正常数据（符合正态分布） ========== + motor_temp = np.random.normal(60, 10, n_samples) # 电机温度：均值60℃，标准差10℃ + battery_temp = np.random.normal(45, 8, n_samples) # 电池温度：均值45℃，标准差8℃ + controller_temp = np.random.normal(50, 9, n_samples) # 控制器温度：均值50℃，标准差9℃ + runtime = np.random.uniform(0, 10, n_samples) # 运行时长：0-10小时均匀分布 + speed = np.random.normal(40, 15, n_samples) # 车速：均值40km/h，标准差15km/h + + # ========== 构造不同等级的异常数据（模拟故障） ========== + # 随机选择30%的样本作为异常样本 anomaly_idx = np.random.choice(n_samples, size=int(n_samples * 0.3), replace=False) - mild_idx = anomaly_idx[:int(len(anomaly_idx) * 0.5)] - moderate_idx = anomaly_idx[int(len(anomaly_idx) * 0.5):int(len(anomaly_idx) * 0.8)] - severe_idx = anomaly_idx[int(len(anomaly_idx) * 0.8):] + # 异常样本细分：50%轻度、30%中度、20%重度 + mild_idx = anomaly_idx[:int(len(anomaly_idx) * 0.5)] # 轻度异常索引 + moderate_idx = anomaly_idx[int(len(anomaly_idx) * 0.5):int(len(anomaly_idx) * 0.8)] # 中度异常索引 + severe_idx = anomaly_idx[int(len(anomaly_idx) * 0.8):] # 重度异常索引 + # 轻度异常：温度小幅升高 motor_temp[mild_idx] += np.random.uniform(5, 10, len(mild_idx)) battery_temp[mild_idx] += np.random.uniform(3, 7, len(mild_idx)) + # 中度异常：温度明显升高（扩展到控制器） motor_temp[moderate_idx] += np.random.uniform(10, 20, len(moderate_idx)) battery_temp[moderate_idx] += np.random.uniform(7, 15, len(moderate_idx)) controller_temp[moderate_idx] += np.random.uniform(8, 12, len(moderate_idx)) + # 重度异常：温度大幅升高 + 运行时长异常 motor_temp[severe_idx] += np.random.uniform(20, 35, len(severe_idx)) battery_temp[severe_idx] += np.random.uniform(15, 25, len(severe_idx)) controller_temp[severe_idx] += np.random.uniform(12, 20, len(severe_idx)) - runtime[severe_idx] = np.random.uniform(8, 12, len(severe_idx)) - - # 生成标签 - labels = np.zeros(n_samples) - labels[mild_idx] = 1 - labels[moderate_idx] = 2 - labels[severe_idx] = 3 - - # 限制数值合理性 - motor_temp = np.clip(motor_temp, 0, 120) - battery_temp = np.clip(battery_temp, 0, 80) - controller_temp = np.clip(controller_temp, 0, 100) - runtime = np.clip(runtime, 0, 12) - speed = np.clip(speed, 0, 120) - + runtime[severe_idx] = np.random.uniform(8, 12, len(severe_idx)) # 运行时长接近上限 + + # ========== 生成标签并限制数值合理性（物理约束） ========== + labels = np.zeros(n_samples) # 初始化标签：0=正常 + labels[mild_idx] = 1 # 轻度异常 + labels[moderate_idx] = 2 # 中度异常 + labels[severe_idx] = 3 # 重度异常 + + # 数值裁剪：确保温度/时长/车速在合理物理范围内 + motor_temp = np.clip(motor_temp, 0, 120) # 电机温度：0-120℃ + battery_temp = np.clip(battery_temp, 0, 80) # 电池温度：0-80℃ + controller_temp = np.clip(controller_temp, 0, 100) # 控制器温度：0-100℃ + runtime = np.clip(runtime, 0, 12) # 运行时长：0-12小时 + speed = np.clip(speed, 0, 120) # 车速：0-120km/h + + # 组合特征矩阵 features = np.column_stack([motor_temp, battery_temp, controller_temp, runtime, speed]) return features, labels -# ===================== 2. 数据预处理 ===================== +# ===================== 2. 数据预处理模块 ===================== def standard_scaler(X): - """手动实现标准化""" - mean = np.mean(X, axis=0) - std = np.std(X, axis=0) - std[std == 0] = 1e-8 + """ + 手动实现标准化（Z-Score）：将特征缩放到均值为0，标准差为1 + 公式：X_scaled = (X - mean) / std + + 参数说明： + -------- + X : np.ndarray + 原始特征矩阵 (n_samples, n_features) + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray, np.ndarray] + X_scaled: 标准化后的特征矩阵 + mean: 各特征的均值 + std: 各特征的标准差（避免除0，替换0为1e-8） + """ + mean = np.mean(X, axis=0) # 按列计算均值 + std = np.std(X, axis=0) # 按列计算标准差 + std[std == 0] = 1e-8 # 防止标准差为0导致除0错误 X_scaled = (X - mean) / std return X_scaled, mean, std def train_test_split(X, y, test_size=0.2, random_state=42): - """手动实现数据集划分""" + """ + 手动实现数据集划分：随机将数据分为训练集和测试集 + 替代sklearn的train_test_split，保证无外部依赖 + + 参数说明： + -------- + X : np.ndarray + 特征矩阵 + y : np.ndarray + 标签数组 + test_size : float + 测试集比例（0-1），默认0.2 + random_state : int + 随机种子（保证划分结果可复现） + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray] + X_train: 训练集特征 + X_test: 测试集特征 + y_train: 训练集标签 + y_test: 测试集标签 + """ np.random.seed(random_state) n_samples = X.shape[0] - test_samples = int(n_samples * test_size) + test_samples = int(n_samples * test_size) # 测试集样本数 + # 随机打乱索引 indices = np.arange(n_samples) np.random.shuffle(indices) + # 划分训练/测试集索引 test_indices = indices[:test_samples] train_indices = indices[test_samples:] + # 按索引划分数据 X_train = X[train_indices] X_test = X[test_indices] y_train = y[train_indices] @@ -85,46 +147,91 @@ def train_test_split(X, y, test_size=0.2, random_state=42): return X_train, X_test, y_train, y_test -# ===================== 3. 决策树分类器 ===================== +# ===================== 3. 决策树分类器（手动实现） ===================== class DecisionTreeClassifier: - """手动实现决策树分类器""" + """ + 手动实现决策树分类器（基于基尼不纯度） + 核心原理：递归划分特征空间，使每个子节点的样本尽可能纯（基尼不纯度最小） + + 参数说明： + -------- + max_depth : int + 决策树最大深度（防止过拟合），默认10 + min_samples_split : int + 节点分裂的最小样本数（防止过拟合），默认5 + """ def __init__(self, max_depth=10, min_samples_split=5): - self.max_depth = max_depth - self.min_samples_split = min_samples_split - self.tree = {} + self.max_depth = max_depth # 树的最大深度 + self.min_samples_split = min_samples_split # 节点分裂最小样本数 + self.tree = {} # 存储决策树结构（字典形式） def _gini_impurity(self, y): - """计算基尼不纯度""" + """ + 计算基尼不纯度（衡量样本纯度） + 公式：Gini = 1 - Σ(p_i²)，p_i为类别i在样本中的占比 + 基尼不纯度越小，样本越纯 + + 参数说明： + -------- + y : np.ndarray + 节点的标签数组 + + 返回值： + -------- + float + 基尼不纯度值（0-1） + """ classes, counts = np.unique(y, return_counts=True) impurity = 1.0 for count in counts: - p = count / len(y) + p = count / len(y) # 类别占比 impurity -= p * p return impurity def _best_split(self, X, y): - """寻找最优划分特征和阈值""" + """ + 寻找最优划分特征和阈值（遍历所有特征和阈值，选择基尼不纯度最小的划分） + + 参数说明： + -------- + X : np.ndarray + 节点的特征矩阵 + y : np.ndarray + 节点的标签数组 + + 返回值： + -------- + Tuple[int, float] + best_feature: 最优划分特征索引 + best_threshold: 最优划分阈值 + """ n_features = X.shape[1] - best_gini = float('inf') - best_feature = None - best_threshold = None + best_gini = float('inf') # 初始化最优基尼不纯度为无穷大 + best_feature = None # 最优特征 + best_threshold = None # 最优阈值 + # 遍历所有特征 for feature in range(n_features): values = X[:, feature] - unique_values = np.unique(values) + unique_values = np.unique(values) # 去重，减少阈值遍历次数 + # 遍历该特征的所有唯一值作为候选阈值 for threshold in unique_values: + # 按阈值划分左右子节点 left_mask = values <= threshold right_mask = values > threshold + # 跳过样本数不足的划分 if len(y[left_mask]) < 1 or len(y[right_mask]) < 1: continue + # 计算划分后的基尼不纯度（加权平均） gini_left = self._gini_impurity(y[left_mask]) gini_right = self._gini_impurity(y[right_mask]) gini = (len(y[left_mask]) * gini_left + len(y[right_mask]) * gini_right) / len(y) + # 更新最优划分 if gini < best_gini: best_gini = gini best_feature = feature @@ -133,27 +240,51 @@ def _best_split(self, X, y): return best_feature, best_threshold def _build_tree(self, X, y, depth=0): - """递归构建决策树""" + """ + 递归构建决策树 + + 参数说明： + -------- + X : np.ndarray + 当前节点的特征矩阵 + y : np.ndarray + 当前节点的标签数组 + depth : int + 当前树的深度 + + 返回值： + -------- + dict/int + 叶子节点：返回类别标签；内部节点：返回划分字典（feature/threshold/left/right） + """ n_samples, n_features = X.shape n_classes = len(np.unique(y)) + # ========== 递归终止条件（叶子节点） ========== + # 1. 树深度达到上限 2. 节点样本数不足 3. 节点样本已纯（单类别） if (depth >= self.max_depth or n_samples < self.min_samples_split or n_classes == 1): + # 返回节点中数量最多的类别 classes, counts = np.unique(y, return_counts=True) return classes[np.argmax(counts)] + # ========== 寻找最优划分并递归构建子树 ========== best_feature, best_threshold = self._best_split(X, y) + # 无有效划分时，返回数量最多的类别 if best_feature is None: classes, counts = np.unique(y, return_counts=True) return classes[np.argmax(counts)] + # 按最优划分拆分数据 left_mask = X[:, best_feature] <= best_threshold right_mask = X[:, best_feature] > best_threshold + # 递归构建左右子树 left_subtree = self._build_tree(X[left_mask], y[left_mask], depth + 1) right_subtree = self._build_tree(X[right_mask], y[right_mask], depth + 1) + # 返回当前节点的划分信息 return { 'feature': best_feature, 'threshold': best_threshold, @@ -162,14 +293,39 @@ def _build_tree(self, X, y, depth=0): } def fit(self, X, y): - """训练决策树""" + """ + 训练决策树（对外接口） + + 参数说明： + -------- + X : np.ndarray + 训练集特征矩阵 + y : np.ndarray + 训练集标签数组 + """ self.tree = self._build_tree(X, y) def _predict_sample(self, x, tree): - """预测单个样本""" + """ + 递归预测单个样本的类别 + + 参数说明： + -------- + x : np.ndarray + 单个样本的特征向量 + tree : dict/int + 决策树结构（递归遍历） + + 返回值： + -------- + int + 样本的预测类别 + """ + # 叶子节点：直接返回类别 if not isinstance(tree, dict): return tree + # 内部节点：按特征和阈值递归 feature = tree['feature'] threshold = tree['threshold'] @@ -179,33 +335,60 @@ def _predict_sample(self, x, tree): return self._predict_sample(x, tree['right']) def predict(self, X): - """预测多个样本""" + """ + 预测多个样本的类别（对外接口） + + 参数说明： + -------- + X : np.ndarray + 待预测的特征矩阵 + + 返回值： + -------- + np.ndarray + 预测类别数组 + """ predictions = [self._predict_sample(x, self.tree) for x in X] return np.array(predictions) -# ===================== 4. 模型评估与可视化 ===================== +# ===================== 4. 模型评估与可视化模块 ===================== def evaluate_model(y_true, y_pred): - """评估模型并生成可视化图表""" + """ + 评估分类模型性能：计算混淆矩阵、精确率、召回率、F1分数，并生成可视化图表 + + 参数说明： + -------- + y_true : np.ndarray + 真实标签数组 + y_pred : np.ndarray + 预测标签数组 + + 返回值： + -------- + np.ndarray + 混淆矩阵 (n_classes, n_classes) + """ classes = np.unique(y_true) n_classes = len(classes) - # 混淆矩阵 + # ========== 1. 计算混淆矩阵 ========== conf_matrix = np.zeros((n_classes, n_classes), dtype=int) for t, p in zip(y_true, y_pred): - conf_matrix[int(t), int(p)] += 1 + conf_matrix[int(t), int(p)] += 1 # 行=真实标签，列=预测标签 - # 计算评估指标 - overall_accuracy = np.trace(conf_matrix) / np.sum(conf_matrix) - precision = [] - recall = [] - f1 = [] + # ========== 2. 计算分类指标 ========== + overall_accuracy = np.trace(conf_matrix) / np.sum(conf_matrix) # 整体准确率（对角线和/总数） + precision = [] # 精确率：TP/(TP+FP) + recall = [] # 召回率：TP/(TP+FN) + f1 = [] # F1分数：2*P*R/(P+R) for i in range(n_classes): - tp = conf_matrix[i, i] - fp = np.sum(conf_matrix[:, i]) - tp - fn = np.sum(conf_matrix[i, :]) - tp + tp = conf_matrix[i, i] # 真正例 + fp = np.sum(conf_matrix[:, i]) - tp # 假正例 + fn = np.sum(conf_matrix[i, :]) - tp # 假负例 + # 计算指标（避免除0） p = tp / (tp + fp) if (tp + fp) > 0 else 0.0 r = tp / (tp + fn) if (tp + fn) > 0 else 0.0 f = 2 * p * r / (p + r) if (p + r) > 0 else 0.0 @@ -214,46 +397,48 @@ def evaluate_model(y_true, y_pred): recall.append(r) f1.append(f) - # 1. 绘制混淆矩阵热力图 + # ========== 3. 可视化：混淆矩阵热力图 ========== plt.figure(figsize=(8, 6)) - plt.imshow(conf_matrix, cmap='Blues') + plt.imshow(conf_matrix, cmap='Blues') # 蓝色系热力图 plt.title('模型混淆矩阵', fontsize=14) plt.xlabel('预测标签', fontsize=12) plt.ylabel('真实标签', fontsize=12) plt.xticks(range(n_classes), ['正常', '轻度异常', '中度异常', '重度异常']) plt.yticks(range(n_classes), ['正常', '轻度异常', '中度异常', '重度异常']) - # 添加数值标注 + # 添加数值标注（显示每个单元格的样本数） for i in range(n_classes): for j in range(n_classes): plt.text(j, i, conf_matrix[i, j], ha='center', va='center', color='black', fontsize=10) plt.colorbar(label='样本数量') plt.tight_layout() - plt.savefig('confusion_matrix.png', dpi=150) # 保存图片 - plt.show() # PyCharm中显示图片 + plt.savefig('confusion_matrix.png', dpi=150) # 保存图片（150DPI） + plt.show() # 显示图片 - # 2. 绘制精确率/召回率/F1对比图 + # ========== 4. 可视化：精确率/召回率/F1对比图 ========== plt.figure(figsize=(10, 6)) x = np.arange(n_classes) - width = 0.25 + width = 0.25 # 柱状图宽度 + # 绘制三组柱状图 plt.bar(x - width, precision, width, label='精确率', color='#1f77b4') plt.bar(x, recall, width, label='召回率', color='#ff7f0e') plt.bar(x + width, f1, width, label='F1分数', color='#2ca02c') + # 图表样式设置 plt.title('模型分类性能指标', fontsize=14) plt.xlabel('异常等级', fontsize=12) plt.ylabel('指标值', fontsize=12) plt.xticks(x, ['正常', '轻度异常', '中度异常', '重度异常']) - plt.ylim(0, 1.1) + plt.ylim(0, 1.1) # Y轴范围0-1.1（便于观察） plt.legend() - plt.grid(axis='y', alpha=0.3) + plt.grid(axis='y', alpha=0.3) # 添加水平网格线 plt.tight_layout() plt.savefig('classification_metrics.png', dpi=150) plt.show() - # 打印文本报告 + # ========== 5. 打印文本报告 ========== print("=== 模型性能报告 ===") print(f"{'类别':<8} {'精确率':<8} {'召回率':<8} {'F1分数':<8}") print("-" * 32) @@ -265,26 +450,37 @@ def evaluate_model(y_true, y_pred): def plot_temperature_distribution(X, y): - """绘制温度特征分布直方图""" + """ + 绘制关键部件温度分布直方图（正常vs异常样本对比） + + 参数说明： + -------- + X : np.ndarray + 特征矩阵 + y : np.ndarray + 标签数组 + """ feature_names = ['电机温度(℃)', '电池温度(℃)', '控制器温度(℃)'] - fig, axes = plt.subplots(1, 3, figsize=(15, 5)) + fig, axes = plt.subplots(1, 3, figsize=(15, 5)) # 1行3列子图 + # 遍历三个温度特征绘制直方图 for i, ax in enumerate(axes): - # 正常数据 - normal_data = X[y == 0, i] - # 异常数据 - anomaly_data = X[y != 0, i] + # 分离正常/异常数据 + normal_data = X[y == 0, i] # 正常样本 + anomaly_data = X[y != 0, i] # 异常样本 + # 绘制密度直方图（density=True：归一化为概率密度） ax.hist(normal_data, bins=30, alpha=0.7, label='正常', color='green', density=True) ax.hist(anomaly_data, bins=30, alpha=0.7, label='异常', color='red', density=True) + # 子图样式设置 ax.set_title(feature_names[i], fontsize=12) ax.set_xlabel('温度值') ax.set_ylabel('概率密度') ax.legend() ax.grid(alpha=0.3) - plt.suptitle('关键部件温度分布对比', fontsize=14) + plt.suptitle('关键部件温度分布对比', fontsize=14) # 总标题 plt.tight_layout() plt.savefig('temperature_distribution.png', dpi=150) plt.show() @@ -292,32 +488,72 @@ def plot_temperature_distribution(X, y): # ===================== 5. 实时监测与报警系统 ===================== class TemperatureAlarmSystem: - """无人车温度报警系统（带可视化）""" + """ + 无人车温度实时报警系统（集成模型预测+硬件阈值+可视化） + 核心功能：模拟传感器数据 → 风险等级预测 → 报警触发 → 数据可视化 + + 参数说明： + -------- + model : DecisionTreeClassifier + 训练好的决策树分类模型 + mean : np.ndarray + 特征均值（标准化用） + std : np.ndarray + 特征标准差（标准化用） + """ def __init__(self, model, mean, std): - self.model = model - self.mean = mean - self.std = std + self.model = model # 分类模型 + self.mean = mean # 标准化均值 + self.std = std # 标准化标准差 + # 报警配置：风险等级→报警信息映射 self.alarm_config = { 0: {"level": "正常", "msg": "各部件温度正常", "action": "持续监测", "color": 'green'}, 1: {"level": "轻度异常", "msg": "温度略高于正常阈值", "action": "提醒驾驶员关注", "color": 'yellow'}, 2: {"level": "中度异常", "msg": "温度明显升高", "action": "降低车速", "color": 'orange'}, 3: {"level": "重度异常", "msg": "温度严重超标", "action": "紧急停车", "color": 'red'} } - self.history = [] + self.history = [] # 存储监测历史数据 + # 硬件安全阈值（直接触发重度报警） self.hard_thresholds = {"motor": 100, "battery": 70, "controller": 90} def _scale_data(self, data): - """标准化数据""" + """ + 标准化传感器数据（与训练数据保持一致） + + 参数说明： + -------- + data : np.ndarray + 原始传感器数据 (1, 5) + + 返回值： + -------- + np.ndarray + 标准化后的数据 + """ scaled = (data - self.mean) / self.std return scaled def _check_hard_threshold(self, sensor_data): - """检查硬件阈值""" + """ + 检查硬件安全阈值（优先级高于模型预测） + 若任一部件温度超过硬件阈值，直接返回重度异常（3） + + 参数说明： + -------- + sensor_data : np.ndarray + 原始传感器数据 (1, 5) + + 返回值： + -------- + int/None + 3=重度异常，None=未触发硬件阈值 + """ motor_temp = sensor_data[0][0] battery_temp = sensor_data[0][1] controller_temp = sensor_data[0][2] + # 检查是否超过硬件阈值 if motor_temp >= self.hard_thresholds["motor"] or \ battery_temp >= self.hard_thresholds["battery"] or \ controller_temp >= self.hard_thresholds["controller"]: @@ -325,27 +561,38 @@ def _check_hard_threshold(self, sensor_data): return None def simulate_sensor(self): - """模拟传感器数据""" + """ + 模拟传感器实时数据生成（含随机异常） + 模拟逻辑：基础正常数据 + 随机概率注入异常 + + 返回值： + -------- + np.ndarray + 传感器数据 (1, 5)：电机、电池、控制器、运行时长、车速 + """ + # 生成基础正常数据 motor = random.uniform(50, 70) battery = random.uniform(40, 50) controller = random.uniform(45, 55) runtime = random.uniform(0, 10) speed = random.uniform(30, 50) + # 随机注入异常 anomaly_prob = random.random() - if anomaly_prob < 0.05: + if anomaly_prob < 0.05: # 5%概率重度异常 motor += 30 battery += 20 controller += 18 - elif anomaly_prob < 0.15: + elif anomaly_prob < 0.15: # 10%概率中度异常 motor += 15 battery += 10 controller += 10 - elif anomaly_prob < 0.3: + elif anomaly_prob < 0.3: # 15%概率轻度异常 motor += 8 battery += 5 controller += 5 + # 数值裁剪（物理约束） motor = max(0, min(motor, 120)) battery = max(0, min(battery, 80)) controller = max(0, min(controller, 100)) @@ -355,20 +602,49 @@ def simulate_sensor(self): return np.array([[motor, battery, controller, runtime, speed]]) def predict_risk(self, sensor_data): - """预测风险等级""" + """ + 预测风险等级（硬件阈值优先 → 模型预测） + + 参数说明： + -------- + sensor_data : np.ndarray + 原始传感器数据 (1, 5) + + 返回值： + -------- + int + 风险等级：0=正常，1=轻度，2=中度，3=重度 + """ + # 第一步：检查硬件阈值（优先级最高） hard_risk = self._check_hard_threshold(sensor_data) if hard_risk is not None: return hard_risk + # 第二步：模型预测（标准化后） scaled_data = self._scale_data(sensor_data) risk_level = self.model.predict(scaled_data)[0] return int(risk_level) def trigger_alarm(self, risk_level, sensor_data): - """触发报警""" + """ + 触发报警并记录日志 + + 参数说明： + -------- + risk_level : int + 风险等级 + sensor_data : np.ndarray + 原始传感器数据 + + 返回值： + -------- + dict + 报警日志 + """ alarm_info = self.alarm_config[risk_level] - timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) # 格式化时间戳 + # 构造日志 log = { "时间": timestamp, "时间戳": time.time(), @@ -382,8 +658,9 @@ def trigger_alarm(self, risk_level, sensor_data): "颜色": alarm_info["color"] } - self.history.append(log) + self.history.append(log) # 保存到历史记录 + # 打印报警信息 print(f"\n【{timestamp}】【{alarm_info['level']}】") print(f"传感器数据：电机{log['电机温度']}℃ | 电池{log['电池温度']}℃ | 控制器{log['控制器温度']}℃") print(f"报警信息：{alarm_info['msg']} | 建议操作：{alarm_info['action']}") @@ -391,12 +668,14 @@ def trigger_alarm(self, risk_level, sensor_data): return log def plot_monitoring_results(self): - """绘制实时监测结果可视化图表""" + """ + 可视化监测结果：温度趋势图 + 报警等级饼图 + """ if not self.history: print("无监测数据可绘制") return - # 提取数据 + # ========== 提取历史数据 ========== timestamps = [log["时间戳"] for log in self.history] motor_temps = [log["电机温度"] for log in self.history] battery_temps = [log["电池温度"] for log in self.history] @@ -404,12 +683,13 @@ def plot_monitoring_results(self): risk_levels = [log["风险等级"] for log in self.history] colors = [log["颜色"] for log in self.history] - # 归一化时间戳（便于显示） + # 归一化时间戳（从0开始） start_ts = timestamps[0] timestamps = [ts - start_ts for ts in timestamps] - # 1. 温度趋势图 + # ========== 1. 温度趋势图 ========== plt.figure(figsize=(12, 6)) + # 绘制温度趋势线 plt.plot(timestamps, motor_temps, 'o-', label='电机温度', linewidth=2, markersize=6) plt.plot(timestamps, battery_temps, 's-', label='电池温度', linewidth=2, markersize=6) plt.plot(timestamps, controller_temps, '^-', label='控制器温度', linewidth=2, markersize=6) @@ -420,6 +700,7 @@ def plot_monitoring_results(self): plt.scatter(t, motor_temps[i], color=colors[i], s=100, edgecolor='black', zorder=5) plt.annotate(f'等级{r}', (t, motor_temps[i]), xytext=(5, 5), textcoords='offset points') + # 图表样式 plt.title('无人车温度实时监测趋势', fontsize=14) plt.xlabel('监测时长(秒)', fontsize=12) plt.ylabel('温度(℃)', fontsize=12) @@ -429,13 +710,14 @@ def plot_monitoring_results(self): plt.savefig('temperature_trend.png', dpi=150) plt.show() - # 2. 报警等级统计饼图 + # ========== 2. 报警等级统计饼图 ========== alarm_counts = defaultdict(int) for log in self.history: alarm_counts[log["报警级别"]] += 1 labels = list(alarm_counts.keys()) sizes = list(alarm_counts.values()) + # 匹配报警颜色 colors = [self.alarm_config[0]["color"] if l == "正常" else self.alarm_config[1]["color"] if l == "轻度异常" else self.alarm_config[2]["color"] if l == "中度异常" else @@ -444,23 +726,33 @@ def plot_monitoring_results(self): plt.figure(figsize=(8, 8)) plt.pie(sizes, labels=labels, colors=colors, autopct='%1.1f%%', startangle=90, shadow=True) plt.title('报警等级分布统计', fontsize=14) - plt.axis('equal') + plt.axis('equal') # 正圆形饼图 plt.tight_layout() plt.savefig('alarm_distribution.png', dpi=150) plt.show() def run_monitor(self, duration=10, interval=2): - """运行实时监测系统""" + """ + 运行实时监测系统 + + 参数说明： + -------- + duration : int + 总监测时长（秒），默认10秒 + interval : int + 采样间隔（秒），默认2秒 + """ print("\n========== 无人车温度报警系统启动 ==========") print(f"监测时长：{duration}秒 | 采样间隔：{interval}秒") print("===========================================\n") start_time = time.time() + # 循环采样直到达到监测时长 while time.time() - start_time < duration: - sensor_data = self.simulate_sensor() - risk_level = self.predict_risk(sensor_data) - self.trigger_alarm(risk_level, sensor_data) - time.sleep(interval) + sensor_data = self.simulate_sensor() # 生成模拟传感器数据 + risk_level = self.predict_risk(sensor_data) # 预测风险等级 + self.trigger_alarm(risk_level, sensor_data) # 触发报警 + time.sleep(interval) # 等待采样间隔 print("\n========== 监测结束 ==========") # 生成监测结果可视化 @@ -481,20 +773,20 @@ def run_monitor(self, duration=10, interval=2): print("=== 1. 生成温度数据集 ===") X, y = generate_temperature_data(n_samples=10000) print(f"数据集规模：特征{X.shape} | 标签{y.shape}") - print(f"标签分布：{np.bincount(y.astype(int))}") + print(f"标签分布：{np.bincount(y.astype(int))}") # 统计各标签数量 - # 绘制温度分布 + # 绘制温度分布对比图 plot_temperature_distribution(X, y) - # 步骤2：数据预处理 + # 步骤2：数据预处理（标准化+划分训练/测试集） print("\n=== 2. 数据预处理 ===") X_scaled, mean, std = standard_scaler(X) X_train, X_test, y_train, y_test = train_test_split(X_scaled, y, test_size=0.2) print(f"训练集：{X_train.shape} | 测试集：{X_test.shape}") - # 步骤3：训练模型 + # 步骤3：训练决策树模型 print("\n=== 3. 训练决策树模型 ===") - model = DecisionTreeClassifier(max_depth=8, min_samples_split=5) + model = DecisionTreeClassifier(max_depth=8, min_samples_split=5) # 调整参数防止过拟合 model.fit(X_train, y_train) print("模型训练完成！") @@ -503,11 +795,12 @@ def run_monitor(self, duration=10, interval=2): y_pred = model.predict(X_test) evaluate_model(y_test, y_pred) - # 步骤5：启动报警系统 + # 步骤5：启动实时报警系统 print("\n=== 5. 启动温度报警系统 ===") alarm_system = TemperatureAlarmSystem(model, mean, std) - alarm_system.run_monitor(duration=10, interval=2) + alarm_system.run_monitor(duration=10, interval=2) # 监测10秒，每2秒采样一次 + # 输出生成的图片列表 print("\n=== 所有可视化图片已保存到当前目录 ===") print("生成的图片：") print("1. temperature_distribution.png - 温度分布对比图") From c31c966f771fdfade30654c261b7eb78f452b976 Mon Sep 17 00:00:00 2001 From: aliangabc Date: Tue, 23 Dec 2025 15:13:34 +0800 Subject: [PATCH 5/8] aaa --- src/unmanned_vehicle_safety_and_operation/README.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 src/unmanned_vehicle_safety_and_operation/README.md diff --git a/src/unmanned_vehicle_safety_and_operation/README.md b/src/unmanned_vehicle_safety_and_operation/README.md new file mode 100644 index 0000000000..67a84f3a35 --- /dev/null +++ b/src/unmanned_vehicle_safety_and_operation/README.md @@ -0,0 +1 @@ +无人车安全与运维系统 \ No newline at end of file From c628c7476aea69cb76ddde5853a2f894bf68ff3f Mon Sep 17 00:00:00 2001 From: aliangabc Date: Tue, 23 Dec 2025 15:19:02 +0800 Subject: [PATCH 6/8] aaa --- src/unmanned_vehicle_safety_and_operation/README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/unmanned_vehicle_safety_and_operation/README.md b/src/unmanned_vehicle_safety_and_operation/README.md index 67a84f3a35..85ee043a15 100644 --- a/src/unmanned_vehicle_safety_and_operation/README.md +++ b/src/unmanned_vehicle_safety_and_operation/README.md @@ -1 +1 @@ -无人车安全与运维系统 \ No newline at end of file +无人车安全与运维 \ No newline at end of file From 8e65f578bdf91c4beeaae52643c0e70e460607b9 Mon Sep 17 00:00:00 2001 From: aliangabc Date: Tue, 23 Dec 2025 18:41:57 +0800 Subject: [PATCH 7/8] aaa --- .../smoke _alarm.py | 314 +++++++++++++----- 1 file changed, 232 insertions(+), 82 deletions(-) diff --git a/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py b/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py index 5388914111..47df1684d3 100644 --- a/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py +++ b/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py @@ -9,87 +9,141 @@ ) import warnings +# 全局设置：忽略无关警告（提升代码运行整洁性） warnings.filterwarnings("ignore") # ===================== 1. 烟雾传感器数据生成（模拟无人车场景） ===================== def generate_smoke_sensor_data( sample_num: int = 1000, # 采样点数（对应时间序列） - sample_freq: int = 1, # 采样频率（Hz） - noise_level: float = 0.05, # 传感器噪声（ppm） - add_abnormal: bool = True # 是否加入烟雾异常值 + sample_freq: int = 1, # 采样频率（Hz），每秒采样次数 + noise_level: float = 0.05, # 传感器噪声幅度（ppm） + add_abnormal: bool = True # 是否加入烟雾异常值（模拟故障/瞬时浓烟） ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: """ - 生成无人车烟雾传感器数据（时间戳+烟雾浓度+报警标签） - 烟雾浓度单位：ppm（百万分之一） - 标签定义：0=无报警（0-5ppm），1=低风险（5-20ppm），2=高风险（>20ppm） + 生成无人车烟雾传感器模拟数据（时间序列），模拟真实车载场景的烟雾浓度变化 + + 核心业务规则： + -------- + - 烟雾浓度单位：ppm（百万分之一） + - 标签定义：0=无报警（0-5ppm），1=低风险（5-20ppm），2=高风险（>20ppm） + - 数据分布：正常样本80%、低风险15%、高风险5%，额外注入突发异常值 + + 参数说明： + -------- + sample_num : int + 总采样点数，默认1000个 + sample_freq : int + 采样频率（Hz），默认1Hz（每秒1个样本） + noise_level : float + 传感器随机噪声幅度（ppm），默认±0.05ppm + add_abnormal : bool + 是否添加突发异常值（模拟传感器故障/瞬时浓烟），默认True + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray, np.ndarray] + time_steps: 时间戳数组（秒），形状=(sample_num,) + smoke_conc: 烟雾浓度数组（ppm），形状=(sample_num,) + labels: 报警标签数组（0/1/2），形状=(sample_num,) """ - # 生成时间戳（模拟连续采样） + # 生成时间戳序列：从0开始，步长=1/采样频率，总长度=采样点数/采样频率 time_steps = np.arange(0, sample_num / sample_freq, 1 / sample_freq) - # 初始化烟雾浓度数组（默认正常浓度） + # 初始化烟雾浓度数组（浮点型，保证精度） smoke_conc = np.zeros(sample_num, dtype=np.float32) - # 生成基础浓度（正常场景：0-5ppm随机波动） + # ========== 1. 生成基础浓度分布 ========== + # 正常场景：0-5ppm随机波动（无报警） base_conc = np.random.uniform(0, 5, sample_num) - # 随机插入低风险/高风险烟雾段（模拟局部烟雾） - # 低风险段（5-20ppm）：占比15% + # ========== 2. 注入风险样本 ========== + # 低风险段（5-20ppm）：占总样本15%，模拟轻微烟雾泄漏 low_risk_idx = np.random.choice(sample_num, int(sample_num * 0.15), replace=False) smoke_conc[low_risk_idx] = np.random.uniform(5, 20, len(low_risk_idx)) - # 高风险段（>20ppm）：占比5% + # 高风险段（>20ppm）：占总样本5%，模拟严重烟雾泄漏 high_risk_idx = np.random.choice(sample_num, int(sample_num * 0.05), replace=False) smoke_conc[high_risk_idx] = np.random.uniform(20, 50, len(high_risk_idx)) - # 正常段赋值 + # 正常段赋值（剩余80%样本） normal_idx = np.setdiff1d(np.arange(sample_num), np.union1d(low_risk_idx, high_risk_idx)) smoke_conc[normal_idx] = base_conc[normal_idx] - # 加入传感器噪声（模拟真实采样误差） + # ========== 3. 加入传感器噪声与物理约束 ========== + # 加入高斯噪声（模拟真实传感器采样误差） smoke_conc += np.random.normal(0, noise_level, sample_num) - smoke_conc = np.maximum(smoke_conc, 0) # 浓度非负 + # 浓度非负约束（物理意义：浓度不能为负） + smoke_conc = np.maximum(smoke_conc, 0) - # 加入突发异常值（模拟传感器故障/瞬时浓烟） + # ========== 4. 注入突发异常值（模拟故障） ========== if add_abnormal: + # 随机生成3-8个异常点，浓度60-100ppm（远超正常范围） abnormal_idx = np.random.choice(sample_num, np.random.randint(3, 8), replace=False) smoke_conc[abnormal_idx] = np.random.uniform(60, 100, len(abnormal_idx)) - # 生成报警标签（基于浓度阈值） - labels = np.zeros(sample_num, dtype=np.int32) - labels[(smoke_conc >= 5) & (smoke_conc < 20)] = 1 # 低风险 - labels[smoke_conc >= 20] = 2 # 高风险 + # ========== 5. 生成报警标签（基于浓度阈值） ========== + labels = np.zeros(sample_num, dtype=np.int32) # 初始化标签为无报警 + labels[(smoke_conc >= 5) & (smoke_conc < 20)] = 1 # 低风险标签 + labels[smoke_conc >= 20] = 2 # 高风险标签 - # 保留2位小数 + # 浓度值保留2位小数（符合传感器实际输出精度） smoke_conc = np.round(smoke_conc, 2) return time_steps, smoke_conc, labels -# ===================== 2. 数据预处理（修复sample_freq未定义问题） ===================== +# ===================== 2. 数据预处理（时序特征工程） ===================== def preprocess_smoke_data(smoke_conc: np.ndarray, labels: np.ndarray, sample_freq: int = 1) -> Tuple[ np.ndarray, np.ndarray, np.ndarray, np.ndarray, StandardScaler, np.ndarray]: """ - 数据预处理：构建时序特征 + 标准化 + 划分训练/测试集 - 特征：当前浓度 + 前1帧浓度 + 浓度变化率（差分） - :param sample_freq: 采样频率（Hz），用于计算浓度变化率 + 烟雾数据预处理：构建时序特征 + 标准化 + 划分训练/测试集 + 核心特征工程：结合时间序列特性，提升模型对浓度变化的感知能力 + + 特征说明： + -------- + - 特征1：当前浓度（ppm） + - 特征2：前1帧浓度（ppm）（首帧补0） + - 特征3：浓度变化率（ppm/s）= 浓度差分 × 采样频率 + + 参数说明： + -------- + smoke_conc : np.ndarray + 烟雾浓度数组（ppm），形状=(n_samples,) + labels : np.ndarray + 报警标签数组，形状=(n_samples,) + sample_freq : int + 采样频率（Hz），用于计算浓度变化率，默认1Hz + + 返回值： + -------- + Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, StandardScaler, np.ndarray] + X_train_scaled: 标准化训练特征，形状=(n_train, 3) + X_test_scaled: 标准化测试特征，形状=(n_test, 3) + y_train: 训练标签，形状=(n_train,) + y_test: 测试标签，形状=(n_test,) + scaler: 拟合后的标准化器（用于实时监测） + features: 全量特征矩阵（未标准化），形状=(n_samples, 3) """ - # 构建时序特征（适配时间序列监测） - # 前1帧浓度（首帧补0） + # ========== 1. 构建时序特征 ========== + # 前1帧浓度（首帧补0，保证特征长度一致） prev_conc = np.concatenate([[0], smoke_conc[:-1]]) - # 浓度变化率（ppm/s）：修复sample_freq参数传递 - conc_diff = np.diff(smoke_conc, prepend=0) * sample_freq # 乘以采样频率转速率 + # 浓度变化率（ppm/s）：差分 × 采样频率（转换为每秒变化量） + conc_diff = np.diff(smoke_conc, prepend=0) * sample_freq - # 特征矩阵（3维特征） + # 组合特征矩阵（3维特征） features = np.column_stack([smoke_conc, prev_conc, conc_diff]) - # 划分训练/测试集（8:2） + # ========== 2. 划分训练/测试集 ========== + # 分层抽样（stratify=labels）：保证训练/测试集标签分布一致 X_train, X_test, y_train, y_test = train_test_split( features, labels, test_size=0.2, random_state=42, stratify=labels ) - # 特征标准化 + # ========== 3. 特征标准化 ========== + # 初始化标准化器（均值0，标准差1） scaler = StandardScaler() + # 训练集拟合+转换，测试集仅转换（避免数据泄露） X_train_scaled = scaler.fit_transform(X_train) X_test_scaled = scaler.transform(X_test) @@ -99,18 +153,31 @@ def preprocess_smoke_data(smoke_conc: np.ndarray, labels: np.ndarray, sample_fre # ===================== 3. 烟雾异常检测（报警触发） ===================== def detect_smoke_abnormal(smoke_conc: np.ndarray, contamination: float = 0.08) -> np.ndarray: """ - 基于孤立森林检测烟雾异常值（触发报警） - :return: 异常标记数组（1=正常，-1=异常/报警） + 基于孤立森林（Isolation Forest）检测烟雾浓度异常值（无监督异常检测） + 核心原理：异常值在特征空间中更容易被孤立，适用于传感器故障/瞬时浓烟检测 + + 参数说明： + -------- + smoke_conc : np.ndarray + 烟雾浓度数组（ppm），形状=(n_samples,) + contamination : float + 异常值占比（0-1），默认0.08（8%） + + 返回值： + -------- + np.ndarray + abnormal_label: 异常标记数组，1=正常，-1=异常/报警，形状=(n_samples,) """ - # 重塑为2D数组（适配sklearn输入） + # 重塑为2D数组（适配sklearn模型输入要求：[n_samples, n_features]） conc_2d = smoke_conc.reshape(-1, 1) - # 训练孤立森林模型（无监督异常检测） + # 初始化孤立森林模型 iso_forest = IsolationForest( - n_estimators=100, - contamination=contamination, # 异常值占比（8%） - random_state=42 + n_estimators=100, # 决策树数量，默认100 + contamination=contamination, # 异常值比例 + random_state=42 # 随机种子（保证结果可复现） ) + # 训练模型并预测异常值 abnormal_label = iso_forest.fit_predict(conc_2d) return abnormal_label @@ -123,17 +190,33 @@ def train_smoke_alarm_model( model_type: str = "random_forest" ) -> object: """ - 训练烟雾报警等级分类模型 + 训练烟雾报警等级分类模型（监督学习），用于预测烟雾风险等级 + + 参数说明： + -------- + X_train : np.ndarray + 标准化训练特征，形状=(n_train, 3) + y_train : np.ndarray + 训练标签，形状=(n_train,) + model_type : str + 模型类型，仅支持"random_forest"，默认随机森林 + + 返回值： + -------- + object + 训练好的分类模型 """ + # 模型选择（当前仅支持随机森林） if model_type == "random_forest": model = RandomForestClassifier( - n_estimators=100, - max_depth=6, - random_state=42 + n_estimators=100, # 决策树数量 + max_depth=6, # 树最大深度（防止过拟合） + random_state=42 # 随机种子 ) else: raise ValueError("仅支持 random_forest 模型") + # 训练模型 model.fit(X_train, y_train) return model @@ -145,31 +228,50 @@ def evaluate_alarm_model( y_test: np.ndarray ) -> Dict[str, float]: """ - 评估报警等级分类lll模型性能 + 评估报警等级分类模型性能，输出分类报告+混淆矩阵可视化 + + 参数说明： + -------- + model : object + 训练好的分类模型 + X_test : np.ndarray + 标准化测试特征，形状=(n_test, 3) + y_test : np.ndarray + 测试标签，形状=(n_test,) + + 返回值： + -------- + Dict[str, float] + 评估指标字典，包含准确率（accuracy） """ # 预测测试集 y_pred = model.predict(X_test) - # 计算准确率 + # 计算整体准确率 accuracy = accuracy_score(y_test, y_pred) - # 输出分类报告 + # 输出详细分类报告（精确率/召回率/F1分数） print("\n===== 报警等级分类报告 =====") print(classification_report( y_test, y_pred, - target_names=["无报警", "低风险", "高风险"] + target_names=["无报警", "低风险", "高风险"] # 标签名称映射 )) - # 绘制混淆矩阵 + # ========== 混淆矩阵可视化 ========== + # 设置中文字体（避免乱码） plt.rcParams["font.sans-serif"] = ["SimHei"] plt.rcParams["axes.unicode_minus"] = False + # 计算混淆矩阵 cm = confusion_matrix(y_test, y_pred) + # 创建画布 fig, ax = plt.subplots(figsize=(8, 6)) + # 绘制混淆矩阵热力图 im = ax.imshow(cm, interpolation='nearest', cmap=plt.cm.Reds) + # 添加颜色条 ax.figure.colorbar(im, ax=ax) - # 设置标签111111 + # 设置坐标轴标签 ax.set( xticks=np.arange(cm.shape[1]), yticks=np.arange(cm.shape[0]), @@ -180,17 +282,19 @@ def evaluate_alarm_model( xlabel="预测等级" ) - # 标注数值 - thresh = cm.max() / 2. + # 在单元格中标注数值 + thresh = cm.max() / 2. # 颜色阈值（区分文字颜色） for i in range(cm.shape[0]): for j in range(cm.shape[1]): ax.text(j, i, format(cm[i, j], 'd'), ha="center", va="center", color="white" if cm[i, j] > thresh else "black") + # 调整布局并显示 fig.tight_layout() plt.show() + # 返回评估指标 return {"accuracy": accuracy} @@ -202,50 +306,69 @@ def visualize_smoke_alarm( alarm_level: np.ndarray ): """ - 可视化烟雾监测结果：浓度曲线 + 异常报警标记 + 等级分类 + 可视化烟雾监测全流程结果： + 1. 浓度时序曲线 + 异常报警标记 + 风险阈值线 + 2. 报警等级分类散点图（按等级着色） + + 参数说明： + -------- + time_steps : np.ndarray + 时间戳数组（秒），形状=(n_samples,) + smoke_conc : np.ndarray + 烟雾浓度数组（ppm），形状=(n_samples,) + abnormal_label : np.ndarray + 异常检测标记数组（1/-1），形状=(n_samples,) + alarm_level : np.ndarray + 报警等级预测数组（0/1/2），形状=(n_samples,) """ + # 设置中文字体 plt.rcParams["font.sans-serif"] = ["SimHei"] plt.rcParams["axes.unicode_minus"] = False - # 创建2行1列子图 + # 创建2行1列子图（总画布大小12×10） fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10)) - # 子图1：烟雾浓度曲线 + 异常报警标记 + # ========== 子图1：烟雾浓度曲线 + 异常报警标记 ========== + # 绘制浓度时序曲线 ax1.plot(time_steps, smoke_conc, color="#1f77b4", linewidth=1.5, label="烟雾浓度") - # 标注异常报警点（红色散点） + # 标注异常报警点（红色散点，zorder=5确保在顶层） alarm_idx = np.where(abnormal_label == -1)[0] ax1.scatter( time_steps[alarm_idx], smoke_conc[alarm_idx], color="#d62728", s=80, label="报警触发点", zorder=5 ) - # 绘制阈值线 + # 绘制风险阈值线（虚线） ax1.axhline(y=5, color="#ff7f0e", linestyle="--", label="低风险阈值（5ppm）") ax1.axhline(y=20, color="#d62728", linestyle="--", label="高风险阈值（20ppm）") + # 子图1样式设置 ax1.set_xlabel("时间 (秒)") ax1.set_ylabel("烟雾浓度 (ppm)") ax1.set_title("无人车烟雾浓度监测曲线 + 报警标记") ax1.legend(loc="upper right") - ax1.grid(alpha=0.3) + ax1.grid(alpha=0.3) # 网格透明度0.3 - # 子图2：报警等级分类结果 + # ========== 子图2：报警等级分类结果 ========== # 等级颜色映射：0=绿色（无报警），1=黄色（低风险），2=红色（高风险） colors = np.array(["#2ca02c", "#ff7f0e", "#d62728"])[alarm_level] + # 绘制等级散点图 ax2.scatter( time_steps, smoke_conc, c=colors, s=20, alpha=0.7, label="报警等级" ) - # 手动添加图例 + # 手动添加图例（解决动态着色无法自动生成图例的问题） ax2.scatter([], [], color="#2ca02c", s=50, label="无报警") ax2.scatter([], [], color="#ff7f0e", s=50, label="低风险") ax2.scatter([], [], color="#d62728", s=50, label="高风险") + # 子图2样式设置 ax2.set_xlabel("时间 (秒)") ax2.set_ylabel("烟雾浓度 (ppm)") ax2.set_title("无人车烟雾报警等级分类") ax2.legend(loc="upper right") ax2.grid(alpha=0.3) + # 调整子图间距并显示 plt.tight_layout() plt.show() @@ -259,84 +382,110 @@ def realtime_smoke_monitor( sample_freq: int = 1 ) -> Tuple[int, str]: """ - 实时监测单样本烟雾浓度，返回报警等级 - :param current_conc: 当前浓度（ppm） - :param prev_conc: 上一帧浓度（ppm） - :return: 等级标签、等级名称 + 车载实时烟雾监测：单样本预测报警等级（适配嵌入式/实时系统） + 核心流程：构建时序特征 → 标准化 → 模型预测 → 等级映射 + + 参数说明： + -------- + model : object + 训练好的分类模型 + scaler : object + 拟合后的标准化器 + current_conc : float + 当前帧烟雾浓度（ppm） + prev_conc : float + 上一帧烟雾浓度（ppm） + sample_freq : int + 采样频率（Hz），用于计算浓度变化率，默认1Hz + + 返回值： + -------- + Tuple[int, str] + level_label: 报警等级标签（0/1/2） + level_name: 报警等级名称（无报警/低风险/高风险） """ - # 计算浓度变化率 + # 计算浓度变化率（ppm/s） conc_diff = (current_conc - prev_conc) * sample_freq - # 构建特征 + # 构建3维特征（适配模型输入） features = np.array([[current_conc, prev_conc, conc_diff]]) - # 标准化 + # 标准化（使用训练好的scaler，避免数据泄露） features_scaled = scaler.transform(features) - # 预测等级 + # 预测等级标签 level_label = model.predict(features_scaled)[0] - # 等级映射 + # 等级名称映射（便于业务理解） level_map = {0: "无报警", 1: "低风险", 2: "高风险"} level_name = level_map[level_label] return level_label, level_name -# ===================== 主函数（修复参数传递） ===================== +# ===================== 主函数（流程整合） ===================== def main(): - # 全局参数 - SAMPLE_NUM = 1000 # 采样点数 + """ + 程序主入口：整合数据生成→异常检测→模型训练→评估→可视化→实时监测全流程 + 模拟无人车烟雾监测系统的完整运行逻辑 + """ + # ========== 全局参数配置 ========== + SAMPLE_NUM = 1000 # 总采样点数 SAMPLE_FREQ = 1 # 采样频率（1Hz） - NOISE_LEVEL = 0.05 # 传感器噪声 + NOISE_LEVEL = 0.05 # 传感器噪声幅度 - # 1. 生成烟雾传感器数据 + # ========== 1. 生成烟雾传感器数据 ========== print("===== 1. 生成无人车烟雾传感器数据 =====") time_steps, smoke_conc, labels = generate_smoke_sensor_data( sample_num=SAMPLE_NUM, sample_freq=SAMPLE_FREQ, noise_level=NOISE_LEVEL ) + # 输出数据基本信息 print(f"生成采样点数：{len(smoke_conc)} 个") print(f"时间范围：0 ~ {time_steps[-1]:.1f} 秒") print(f"浓度范围：{smoke_conc.min():.2f} ~ {smoke_conc.max():.2f} ppm") - # 2. 烟雾异常检测（触发报警） + # ========== 2. 烟雾异常检测（报警触发） ========== print("\n===== 2. 烟雾异常检测（报警触发） =====") abnormal_label = detect_smoke_abnormal(smoke_conc) alarm_count = len(np.where(abnormal_label == -1)[0]) print(f"检测到报警触发点数量：{alarm_count} 个") - alarm_time = time_steps[abnormal_label == -1][:5] # 打印前5个报警时间 + # 输出前5个报警时间戳（示例） + alarm_time = time_steps[abnormal_label == -1][:5] print(f"前5个报警时间戳：{alarm_time.round(2)} 秒") - # 3. 数据预处理（修复：传递sample_freq参数） + # ========== 3. 数据预处理 ========== print("\n===== 3. 数据预处理 =====") X_train, X_test, y_train, y_test, scaler, features = preprocess_smoke_data( - smoke_conc, labels, sample_freq=SAMPLE_FREQ # 补充sample_freq参数 + smoke_conc, labels, sample_freq=SAMPLE_FREQ ) print(f"训练集数量：{len(X_train)} 条，测试集数量：{len(X_test)} 条") - # 4. 训练报警等级分类模型 + # ========== 4. 训练报警等级分类模型 ========== print("\n===== 4. 训练烟雾报警等级模型 =====") alarm_model = train_smoke_alarm_model(X_train, y_train) - # 5. 评估模型 + # ========== 5. 模型评估 ========== print("\n===== 5. 模型评估 =====") eval_metrics = evaluate_alarm_model(alarm_model, X_test, y_test) print(f"报警等级分类准确率：{eval_metrics['accuracy']:.2%}") - # 6. 全量数据等级预测 + # ========== 6. 全量数据等级预测 ========== print("\n===== 6. 全量数据报警等级预测 =====") + # 全量特征标准化 features_scaled = scaler.transform(features) + # 预测全量数据等级 alarm_level = alarm_model.predict(features_scaled) + # 统计各等级样本数 level_count = {0: 0, 1: 0, 2: 0} for level in alarm_level: level_count[level] += 1 print(f"无报警样本数：{level_count[0]} | 低风险样本数：{level_count[1]} | 高风险样本数：{level_count[2]}") - # 7. 可视化监测结果 222 + # ========== 7. 可视化监测结果 ========== print("\n===== 7. 可视化烟雾监测结果 =====") visualize_smoke_alarm(time_steps, smoke_conc, abnormal_label, alarm_level) - # 8. 实时监测示例（模拟车载传感器） + # ========== 8. 实时监测示例（模拟车载场景） ========== print("\n===== 8. 实时烟雾监测示例 =====") # 示例1：正常浓度（3ppm） level1, name1 = realtime_smoke_monitor(alarm_model, scaler, 3.0, 2.8, SAMPLE_FREQ) @@ -351,5 +500,6 @@ def main(): print(f"实时样本3：当前浓度25.0ppm → 报警等级：{name3}") +# 程序执行入口 if __name__ == "__main__": main() \ No newline at end of file From 63b04a41e403483ebafd5209b1267068be3d1bc9 Mon Sep 17 00:00:00 2001 From: aliangabc Date: Sat, 27 Dec 2025 17:20:33 +0800 Subject: [PATCH 8/8] aaa --- src/unmanned_vehicle_safety_and_operation/smoke _alarm.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py b/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py index 47df1684d3..c73fe1f020 100644 --- a/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py +++ b/src/unmanned_vehicle_safety_and_operation/smoke _alarm.py @@ -491,7 +491,7 @@ def main(): level1, name1 = realtime_smoke_monitor(alarm_model, scaler, 3.0, 2.8, SAMPLE_FREQ) print(f"实时样本1：当前浓度3.0ppm → 报警等级：{name1}") - # 示例2：低风险（10ppm） + # 示例2：低风险（10ppm）111 level2, name2 = realtime_smoke_monitor(alarm_model, scaler, 10.0, 8.5, SAMPLE_FREQ) print(f"实时样本2：当前浓度10.0ppm → 报警等级：{name2}")