import numpy as np import pandas as pd import calc_way from scipy import stats import calc_slope_line import matplotlib.pyplot as plt import model import os import numpy as np import matplotlib.pyplot as plt from sklearn.linear_model import RANSACRegressor, LinearRegression from sklearn.cluster import DBSCAN # 数据截断线 model = model.Model() # limit_slope = model.limit_slope # limit_intercept = model.limit_intercept def grid_downsample(points, cell_size=15): """网格化降采样，保持空间结构""" df = pd.DataFrame(points, columns=['x', 'y']) df['x_grid'] = (df['x'] // cell_size) * cell_size df['y_grid'] = (df['y'] // cell_size) * cell_size sampled = df.groupby(['x_grid', 'y_grid']).first().reset_index() return sampled[['x', 'y']].values """ 读取yolo网络识别路沿的坐标数据,筛选出目标区域的数据点，并将路沿上下侧数据分离 参数：保存数据的txt文件路径 返回值：在目标区域内的下侧数据点坐标x_bot、y_bot，上侧数据点坐标x_top,y_top """ def get_data(txt_name,jingdu): # 加载数据 data = np.loadtxt(txt_name) int_data = data.astype(int) # 网格化降采样 grid_sampled = grid_downsample(int_data, cell_size=20) # 数据截断 x = [] y = [] for i in range(grid_sampled.shape[0]): grid_sampled[i][1] = 960 - int(grid_sampled[i][1]) if limit_slope * int(grid_sampled[i][0]) + limit_intercept - int(grid_sampled[i][1]) < 0: continue x.append(int(grid_sampled[i][0])) y.append(int(grid_sampled[i][1])) x = np.array(x) y = np.array(y) # 原始数据粗分类 slope, intercept, r_2 = calc_slope_line.linear_regression(x, y) y_pred = slope * x + intercept x_bot = [] y_bot = [] x_top = [] y_top = [] for i in range(len(x)): if x[i] * slope + intercept - y[i] > 0: x_bot.append(x[i]) y_bot.append(y[i]) else: x_top.append(x[i]) y_top.append(y[i]) x_bot = np.array(x_bot) y_bot = np.array(y_bot) x_top = np.array(x_top) y_top = np.array(y_top) slope_bot, intercept_bot, r2_bot = calc_slope_line.linear_regression(x_bot, y_bot) slope_top, intercept_top, r2_top = calc_slope_line.linear_regression(x_top, y_top) print(f"未清洗数据拟合上下沿：r2_bot = {r2_bot},r2_top = {r2_top}") # 第一次数据清洗，消除误识别点 # 计算残差 residuals = y - y_pred # 计算残差的标准差 (MSE的平方根) residual_std = np.sqrt(np.sum(residuals ** 2) / (len(x) - 2)) standardized_residuals = residuals / residual_std # 设置阈值 (常用 2.5-3.0 个标准差) threshold = 2.0 # 标记异常点 outlier_mask = np.abs(standardized_residuals) > threshold outliers_x = x[outlier_mask] outliers_y = y[outlier_mask] print(f"第一次数据清洗发现 {np.sum(outlier_mask)} 个异常点:") for i, (x_val, y_val) in enumerate(zip(outliers_x, outliers_y)): print(f"点 {i + 1}: x={x_val}, y={y_val}, 残差={residuals[outlier_mask][i]:.2f}") # 剔除异常点 clean_x = x[~outlier_mask] clean_y = y[~outlier_mask] clean_slope, clean_intercept, clean_r_2 = calc_slope_line.linear_regression(clean_x, clean_y) print(f"清洗数据后整体拟合参数r_2 = {r_2}") # 第一次数据清洗后的数据再分类 x_bot_clean = [] y_bot_clean = [] x_top_clean = [] y_top_clean = [] for i in range(len(clean_x)): if clean_x[i] * clean_slope + clean_intercept - clean_y[i] > 0: x_bot_clean.append(clean_x[i]) y_bot_clean.append(clean_y[i]) else: x_top_clean.append(clean_x[i]) y_top_clean.append(clean_y[i]) x_bot_clean = np.array(x_bot_clean) y_bot_clean = np.array(y_bot_clean) x_top_clean = np.array(x_top_clean) y_top_clean = np.array(y_top_clean) # 第二次数据清洗，消除误分类点 clean_slope_bot, clean_intercept_bot, clean_r2_bot = calc_slope_line.linear_regression(x_bot_clean, y_bot_clean) clean_slope_top, clean_intercept_top, clean_r2_top = calc_slope_line.linear_regression(x_top_clean, y_top_clean) print(f"清洗数据后上下沿拟合参数clean_r2_bot = {clean_r2_bot},clean_r2_top = {clean_r2_top}") # 绘制拟合线 y_bot_pred = clean_slope_bot * x_bot_clean + clean_intercept_bot y_top_pred = clean_slope_top * x_top_clean + clean_intercept_top # 计算残差 residuals_bot = y_bot_clean - y_bot_pred residuals_top = y_top_clean - y_top_pred # 计算残差的标准差 (MSE的平方根) residual_std_bot = np.sqrt(np.sum(residuals_bot ** 2) / (len(x_bot_clean) - 2)) residual_std_top = np.sqrt(np.sum(residuals_top ** 2) / (len(x_top_clean) - 2)) # 计算标准化残差 (Z-score) standardized_residuals_bot = residuals_bot / residual_std_bot standardized_residuals_top = residuals_top / residual_std_top # 设置阈值 (常用 2.5-3.0 个标准差) threshold = 1.5 # 标记异常点 outlier_mask_bot = np.abs(standardized_residuals_bot) > threshold outlier_mask_top = np.abs(standardized_residuals_top) > threshold outliers_x_bot = x_bot_clean[outlier_mask_bot] outliers_y_bot = y_bot_clean[outlier_mask_bot] outliers_x_top = x_top_clean[outlier_mask_top] outliers_y_top = y_top_clean[outlier_mask_top] print(f"第二次数据清洗下沿发现 {np.sum(outlier_mask_bot)} 个异常点:") # for i, (x_val, y_val) in enumerate(zip(outliers_x_bot, outliers_y_bot)): # print(f"点 {i + 1}: x={x_val}, y={y_val}, 残差={residuals_bot[outlier_mask_bot][i]:.2f}") print(f"第二次数据清洗上沿发现 {np.sum(outlier_mask_top)} 个异常点:") # for i, (x_val, y_val) in enumerate(zip(outliers_x_top, outliers_y_top)): # print(f"点 {i + 1}: x={x_val}, y={y_val}, 残差={residuals_top[outlier_mask_top][i]:.2f}") # 剔除异常点 x_bot_clean = x_bot_clean[~outlier_mask_bot] y_bot_clean = y_bot_clean[~outlier_mask_bot] x_top_clean = x_top_clean[~outlier_mask_top] y_top_clean = y_top_clean[~outlier_mask_top] # 判断数据的有效性 clean_slope_bot, clean_intercept_bot, clean_r2_bot = calc_slope_line.linear_regression(x_bot_clean, y_bot_clean) clean_slope_top, clean_intercept_top, clean_r2_top = calc_slope_line.linear_regression(x_top_clean, y_top_clean) print(f"清洗数据后上下沿拟合参数clean_r2_bot = {clean_r2_bot},clean_r2_top = {clean_r2_top}") if ((1-clean_r2_bot) > (1-jingdu)) or ((1-clean_r2_top) > (1-jingdu)): print("无效数据") return 0, None, None, None, None return 1, x_bot_clean, y_bot_clean, x_top_clean, y_top_clean def filter_middle_80_percent(data): """ 保留数组中间80%的数据（删除首尾各10%）。 参数: data (np.ndarray): 输入数组（可以是一维或多维，但会先展平）。 返回: np.ndarray: 中间80%的数据。 """ # 展平数组（确保处理的是所有数据点） flattened_data = data.flatten() # # 计算10%和90%分位数 # lower_bound = np.percentile(flattened_data, 15) # upper_bound = np.percentile(flattened_data, 75) # # # 筛选中间80%的数据 # mask = (data >= lower_bound) & (data <= upper_bound) # 计算最大值、最小值和总范围 data_min = np.min(flattened_data) data_max = np.max(flattened_data) data_range = data_max - data_min # 计算中间80%的上下界 lower_bound = data_min + 0.2 * data_range upper_bound = data_max - 0.1 * data_range # 筛选数据 mask = (flattened_data >= lower_bound) & (flattened_data <= upper_bound) return mask def load_data(txt_name): """从用户输入的文件路径加载二维XY数据""" while True: filepath = txt_name.strip() if filepath.lower() == 'q': return None, None try: data = np.loadtxt(filepath) if data.shape[1] != 2: print("错误：文件必须包含两列数据（X和Y）") continue x = data[:, 0].reshape(-1, 1) y = data[:, 1].reshape(-1, 1) y = 960 - y mask = filter_middle_80_percent(x) x_clean = x[mask] y_clean = y[mask] return x_clean.reshape(-1,1), y_clean.reshape(-1,1) except Exception as e: print(f"加载文件出错: {e}") def ransac_fit(x, y, residual_threshold=2.0): """执行RANSAC拟合并返回模型和内点/外点""" ransac = RANSACRegressor( LinearRegression(), residual_threshold=residual_threshold, random_state=42 ) ransac.fit(x, y) inlier_mask = ransac.inlier_mask_ outlier_mask = ~inlier_mask return ransac.estimator_, inlier_mask, outlier_mask def get_data(txt_name): # print("=== 双重RANSAC拟合与异常值聚类分析 ===") # 加载数据 x, y = load_data(txt_name) if x is None: return 0, None, None, None, None, None, None, None, None # 第一次RANSAC拟合 # print("\n正在进行第一次RANSAC拟合...") model1, inlier_mask1, outlier_mask1 = ransac_fit(x, y, residual_threshold=3.0) x_inliers1 = x[inlier_mask1] y_inliers1 = y[inlier_mask1] # 获取第一次拟合的外点 x_outliers1 = x[outlier_mask1] y_outliers1 = y[outlier_mask1] # 第二次RANSAC拟合（在外点上） model2, inlier_mask2, outlier_mask2 = None, None, None if len(x_outliers1) > 10: # 确保有足够的外点进行第二次拟合 # print("\n正在进行第二次RANSAC拟合...") model2, inlier_mask2, outlier_mask2 = ransac_fit(x_outliers1, y_outliers1, residual_threshold=3.0) x_inliers2 = x_outliers1[inlier_mask2] y_inliers2 = y_outliers1[inlier_mask2] # 获取第二次拟合的外点 x_outliers2 = x_outliers1[outlier_mask2] y_outliers2 = y_outliers1[outlier_mask2] mean_outliers1 = np.mean(y_inliers1) mean_outliers2 = np.mean(y_inliers2) m1 = model1.predict(np.array([600]).reshape(-1, 1)) m2 = model2.predict(np.array([600]).reshape(-1, 1)) # 判断上下沿 if m1 > m2: model_top, model_bot = model1, model2 x_top, x_bot = x_inliers1, x_inliers2 y_top, y_bot = y_inliers1, y_inliers2 else: model_top, model_bot = model2, model1 x_top, x_bot = x_inliers2, x_inliers1 y_top, y_bot = y_inliers2, y_inliers1 # 统一提取斜率和截距 slope_top = model_top.coef_[0][0] intercept_top = model_top.intercept_[0] slope_bot = model_bot.coef_[0][0] intercept_bot = model_bot.intercept_[0] plt.figure(figsize=(14, 7)) # 绘制原始内点 plt.scatter(x_bot, y_bot, color='limegreen', marker='o', s=30, alpha=0.7, label='bot') plt.scatter(x_top, y_top, color='red', marker='*', s=100, edgecolor='black', label='top') plt.xlabel('X', fontsize=12) plt.ylabel('Y', fontsize=12) plt.title(f'{txt_name}', fontsize=14) plt.legend(fontsize=10, loc='best') plt.grid(True, alpha=0.3) # plt.tight_layout() plt.show() # print(f"model_top = {model_top.coef_, model_top.intercept_}") # print("\n分析完成！") return 1, x_bot, y_bot, slope_bot, intercept_bot, x_top, y_top, slope_top, intercept_top