当前位置：首页 > news >正文

基于深度学习的双对数坐标曲线转折点识别方法研究

news 2025/9/24 13:17:14

基于深度学习的双对数坐标曲线转折点识别方法研究

1. 引言

1.1 研究背景与意义

在科学研究和工程应用领域，双对数坐标曲线广泛应用于数据分析与可视化。许多自然现象和工程问题中的变量关系在双对数坐标系下呈现出线性或分段线性特征，如材料力学中的应力-应变关系、地球物理学中的地震波分析、经济学中的规模效应研究等。这些曲线中的转折点往往对应着重要的物理机制转变或系统状态变化，因此准确识别转折点对于理解现象本质具有重要意义。

传统的转折点识别方法主要基于数值微分、曲率分析或分段拟合等技术，但这些方法在面对噪声干扰、数据稀疏或复杂形态的曲线时往往表现不佳。近年来，深度学习技术在图像识别、信号处理等领域取得了显著成功，为曲线转折点识别提供了新的解决方案。

1.2 研究现状

目前，针对曲线转折点识别的研究主要集中在以下几个方向：

基于数值分析的方法：通过计算曲线的一阶或二阶导数来识别斜率变化点，但对噪声敏感且需要平滑预处理。
基于统计模型的方法：使用回归分析或变点检测算法，如贝叶斯变点检测、CUSUM算法等，但对曲线形态假设较强。
基于机器学习的方法：应用支持向量机、随机森林等传统机器学习算法，需要手动设计特征且泛化能力有限。
基于深度学习的方法：利用神经网络自动学习曲线特征，逐渐成为研究热点，但在双对数坐标曲线领域的应用尚不充分。

1.3 本文研究内容

本文旨在开发一种基于深度学习的双对数坐标曲线转折点批量识别方法，主要研究内容包括：

双对数坐标曲线转折点的数学定义与特征分析
适用于转折点识别的深度学习模型架构设计
大规模双对数曲线数据集的构建与增强方法
模型训练策略与优化算法研究
转折点识别系统的实现与性能评估

2. 双对数坐标曲线转折点理论基础

2.1 双对数坐标系统特性

双对数坐标系是指两个坐标轴都采用对数刻度的坐标系系统。在这种坐标系下，幂函数关系 $y = ax^b$ 将呈现为线性关系：

$log⁡(y)=log⁡(a)+blog⁡(x)\log(y) = \log(a) + b\log(x)$

这一特性使得双对数坐标在分析幂律关系、指数增长等现象时具有独特优势。曲线在双对数坐标下的转折点通常表示系统主导机制的变化或不同物理过程的过渡。

2.2 转折点的数学定义

在双对数坐标下，转折点可以定义为曲线曲率发生显著变化的点。数学上，可以通过以下方式定义：

设曲线在双对数坐标下表示为 $Y = F (X)$ ，其中 $\log(x)$ , $\log(y)$ 。则曲线的一阶导数 $F^{'} (X)$ 表示局部斜率，二阶导数 $F^{''} (X)$ 表示曲率变化。

转折点判定条件：

曲率极值点： $∣ F^{''} (X) ∣$ 取得局部极大值
斜率变化显著： $∣F′(X+)−F′(X−)∣>Δthreshold|F'(X^+) - F'(X^-)| > \Delta_{threshold}$
统计显著性：转折点前后区域的统计特性有显著差异

2.3 转折点的物理意义

在不同应用领域中，双对数坐标曲线转折点对应着不同的物理意义：

材料科学：应力-应变曲线转折点可能表示材料从弹性变形到塑性变形的转变
地球物理：地震波谱转折点可能反映不同地层结构的界面
流体力学：流速分布转折点可能指示流动状态的转变（如层流到湍流）
经济学：生产函数转折点可能表示规模收益的变化

3. 深度学习模型设计

3.1 模型选择依据

针对双对数坐标曲线转折点识别任务，我们考虑以下深度学习模型选项：

卷积神经网络(CNN)：适合提取曲线的局部形态特征
循环神经网络(RNN)：适合处理序列数据，捕捉前后文关系
注意力机制模型：可聚焦于曲线的关键区域
图神经网络(GNN)：将曲线视为图结构处理

综合分析任务特点，我们选择以CNN为主干网络，结合注意力机制的混合架构，既能捕捉局部特征又能关注全局上下文。

3.2 模型架构设计

我们设计了一个多尺度特征融合的深度学习模型，具体架构如下：

3.2.1 输入层与预处理

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as npclass CurvePreprocessor(nn.Module):"""曲线预处理模块：将原始曲线数据转换为模型输入格式"""def __init__(self, input_size=1024, output_size=512):super(CurvePreprocessor, self).__init__()self.input_size = input_sizeself.output_size = output_sizedef forward(self, x):# x: [batch_size, 2, input_size] - 2表示(X,Y)坐标# 标准化处理x_normalized = self.normalize_curve(x)# 插值到固定长度if x_normalized.size(2) != self.output_size:x_resampled = F.interpolate(x_normalized, size=self.output_size, mode='linear', align_corners=False)else:x_resampled = x_normalizedreturn x_resampleddef normalize_curve(self, x):# 对每个曲线的X和Y分别进行标准化batch_size, _, seq_len = x.shape# 分离X和Y坐标x_coord = x[:, 0:1, :]  # [batch_size, 1, seq_len]y_coord = x[:, 1:2, :]  # [batch_size, 1, seq_len]# 对X坐标进行min-max标准化x_min = x_coord.min(dim=2, keepdim=True)[0]x_max = x_coord.max(dim=2, keepdim=True)[0]x_normalized = (x_coord - x_min) / (x_max - x_min + 1e-8)# 对Y坐标进行z-score标准化y_mean = y_coord.mean(dim=2, keepdim=True)y_std = y_coord.std(dim=2, keepdim=True)y_normalized = (y_coord - y_mean) / (y_std + 1e-8)# 合并标准化后的坐标normalized = torch.cat([x_normalized, y_normalized], dim=1)return normalized

3.2.2 多尺度特征提取模块

class MultiScaleFeatureExtractor(nn.Module):"""多尺度特征提取模块：使用不同大小的卷积核捕捉不同尺度的曲线特征"""def __init__(self, input_channels=2, feature_dim=64):super(MultiScaleFeatureExtractor, self).__init__()# 小尺度特征提取（局部细节）self.conv_small = nn.Sequential(nn.Conv1d(input_channels, feature_dim//4, kernel_size=3, padding=1),nn.BatchNorm1d(feature_dim//4),nn.ReLU(inplace=True),nn.Conv1d(feature_dim//4, feature_dim//4, kernel_size=3, padding=1),nn.BatchNorm1d(feature_dim//4),nn.ReLU(inplace=True),)# 中尺度特征提取self.conv_medium = nn.Sequential(nn.Conv1d(input_channels, feature_dim//2, kernel_size=7, padding=3),nn.BatchNorm1d(feature_dim//2),nn.ReLU(inplace=True),nn.Conv1d(feature_dim//2, feature_dim//2, kernel_size=7, padding=3),nn.BatchNorm1d(feature_dim//2),nn.ReLU(inplace=True),)# 大尺度特征提取（全局形态）self.conv_large = nn.Sequential(nn.Conv1d(input_channels, feature_dim, kernel_size=15, padding=7),nn.BatchNorm1d(feature_dim),nn.ReLU(inplace=True),nn.Conv1d(feature_dim, feature_dim, kernel_size=15, padding=7),nn.BatchNorm1d(feature_dim),nn.ReLU(inplace=True),)# 特征融合self.feature_fusion = nn.Sequential(nn.Conv1d(feature_dim + feature_dim//2 + feature_dim//4, feature_dim, kernel_size=1),nn.BatchNorm1d(feature_dim),nn.ReLU(inplace=True),)def forward(self, x):# 提取多尺度特征feat_small = self.conv_small(x)feat_medium = self.conv_medium(x)feat_large = self.conv_large(x)# 调整特征维度一致feat_medium = F.interpolate(feat_medium, size=feat_small.size(2), mode='linear', align_corners=False)feat_large = F.interpolate(feat_large, size=feat_small.size(2), mode='linear', align_corners=False)# 特征拼接与融合fused_feat = torch.cat([feat_small, feat_medium, feat_large], dim=1)output = self.feature_fusion(fused_feat)return output

3.2.3 注意力机制模块

class CurveAttention(nn.Module):"""曲线注意力机制：聚焦于曲线的关键区域"""def __init__(self, feature_dim, num_heads=4):super(CurveAttention, self).__init__()self.feature_dim = feature_dimself.num_heads = num_heads# 自注意力机制self.self_attention = nn.MultiheadAttention(embed_dim=feature_dim, num_heads=num_heads,batch_first=True)# 位置编码self.position_encoding = nn.Parameter(torch.randn(1, 1000, feature_dim)  # 最大序列长度1000)# 前馈网络self.feed_forward = nn.Sequential(nn.Linear(feature_dim, feature_dim * 2),nn.ReLU(inplace=True),nn.Linear(feature_dim * 2, feature_dim),)self.layer_norm1 = nn.LayerNorm(feature_dim)self.layer_norm2 = nn.LayerNorm(feature_dim)def forward(self, x):# x: [batch_size, feature_dim, seq_len]batch_size, feature_dim, seq_len = x.shape# 转换维度: [batch_size, seq_len, feature_dim]x_transposed = x.transpose(1, 2)# 添加位置编码if seq_len <= 1000:pos_enc = self.position_encoding[:, :seq_len, :]x_with_pos = x_transposed + pos_encelse:# 如果序列过长，进行插值pos_enc = F.interpolate(self.position_encoding.transpose(1, 2), size=seq_len, mode='linear', align_corners=False).transpose(1, 2)x_with_pos = x_transposed + pos_enc# 自注意力attn_output, attn_weights = self.self_attention(x_with_pos, x_with_pos, x_with_pos)# 残差连接与层归一化x_norm1 = self.layer_norm1(x_with_pos + attn_output)# 前馈网络ff_output = self.feed_forward(x_norm1)# 残差连接与层归一化x_norm2 = self.layer_norm2(x_norm1 + ff_output)# 恢复原始维度: [batch_size, feature_dim, seq_len]output = x_norm2.transpose(1, 2)return output, attn_weights

3.2.4 转折点检测头

class TurningPointHead(nn.Module):"""转折点检测头：预测每个点是否为转折点的概率"""def __init__(self, feature_dim, num_classes=3):super(TurningPointHead, self).__init__()# num_classes: 0-非转折点, 1-弱转折点, 2-强转折点self.detection_head = nn.Sequential(nn.Conv1d(feature_dim, feature_dim//2, kernel_size=3, padding=1),nn.BatchNorm1d(feature_dim//2),nn.ReLU(inplace=True),nn.Conv1d(feature_dim//2, feature_dim//4, kernel_size=3, padding=1),nn.BatchNorm1d(feature_dim//4),nn.ReLU(inplace=True),nn.Conv1d(feature_dim//4, num_classes, kernel_size=1),)# 置信度预测self.confidence_head = nn.Sequential(nn.AdaptiveAvgPool1d(1),  # 全局平均池化nn.Flatten(),nn.Linear(feature_dim, feature_dim//2),nn.ReLU(inplace=True),nn.Linear(feature_dim//2, 1),nn.Sigmoid())def forward(self, x):# 转折点分类point_predictions = self.detection_head(x)  # [batch_size, num_classes, seq_len]# 置信度预测confidence = self.confidence_head(x)  # [batch_size, 1]return point_predictions, confidence

3.2.5 完整模型集成

class LogLogTurningPointDetector(nn.Module):"""完整的双对数坐标曲线转折点检测模型"""def __init__(self, input_size=1024, feature_dim=128, num_classes=3):super(LogLogTurningPointDetector, self).__init__()self.preprocessor = CurvePreprocessor(input_size, 512)self.feature_extractor = MultiScaleFeatureExtractor(2, feature_dim)self.attention = CurveAttention(feature_dim)self.detection_head = TurningPointHead(feature_dim, num_classes)# 深度监督分支self.auxiliary_head = nn.Sequential(nn.Conv1d(feature_dim, num_classes, kernel_size=1),)def forward(self, x, use_attention=True):# 预处理x_processed = self.preprocessor(x)# 特征提取features = self.feature_extractor(x_processed)# 注意力机制（可选）if use_attention:features, attn_weights = self.attention(features)else:attn_weights = None# 主检测头main_predictions, confidence = self.detection_head(features)# 辅助头（用于深度监督）auxiliary_predictions = self.auxiliary_head(features)return {'main_predictions': main_predictions,'auxiliary_predictions': auxiliary_predictions,'confidence': confidence,'attention_weights': attn_weights}

3.3 损失函数设计

转折点检测任务需要特殊的损失函数设计，以应对类别不平衡和多尺度检测的挑战：

class TurningPointLoss(nn.Module):"""转折点检测的复合损失函数"""def __init__(self, alpha=0.25, gamma=2.0, lambda_cls=1.0, lambda_aux=0.5, lambda_conf=0.1):super(TurningPointLoss, self).__init__()self.alpha = alphaself.gamma = gammaself.lambda_cls = lambda_clsself.lambda_aux = lambda_auxself.lambda_conf = lambda_conf# 焦点损失解决类别不平衡self.focal_loss = FocalLoss(alpha=alpha, gamma=gamma)# 平滑L1损失用于位置回归self.smooth_l1 = nn.SmoothL1Loss()def forward(self, predictions, targets):"""predictions: 模型输出字典targets: 包含真实标签的字典"""main_pred = predictions['main_predictions']aux_pred = predictions['auxiliary_predictions']confidence = predictions['confidence']# 分类损失（焦点损失）cls_loss_main = self.focal_loss(main_pred, targets['point_labels'])cls_loss_aux = self.focal_loss(aux_pred, targets['point_labels'])# 置信度损失conf_loss = F.binary_cross_entropy(confidence, targets['curve_confidence'])# 总损失total_loss = (self.lambda_cls * cls_loss_main + self.lambda_aux * cls_loss_aux + self.lambda_conf * conf_loss)return {'total_loss': total_loss,'cls_loss_main': cls_loss_main,'cls_loss_aux': cls_loss_aux,'conf_loss': conf_loss}class FocalLoss(nn.Module):"""焦点损失：解决类别不平衡问题"""def __init__(self, alpha=0.25, gamma=2.0, reduction='mean'):super(FocalLoss, self).__init__()self.alpha = alphaself.gamma = gammaself.reduction = reductiondef forward(self, inputs, targets):BCE_loss = F.cross_entropy(inputs, targets, reduction='none')pt = torch.exp(-BCE_loss)  # 防止NaNF_loss = self.alpha * (1-pt)**self.gamma * BCE_lossif self.reduction == 'mean':return torch.mean(F_loss)elif self.reduction == 'sum':return torch.sum(F_loss)else:return F_loss

4. 数据准备与增强

4.1 双对数曲线数据生成

由于真实双对数曲线数据获取困难，我们设计了综合的数据生成策略：

import numpy as np
from scipy import interpolate
import matplotlib.pyplot as pltclass LogLogCurveGenerator:"""双对数坐标曲线生成器"""def __init__(self, num_points_range=(100, 1000), noise_level=0.1):self.num_points_range = num_points_rangeself.noise_level = noise_leveldef generate_piecewise_power_law(self, num_segments=2):"""生成分段幂律曲线模拟真实双对数坐标数据"""# 生成分段点breakpoints = sorted(np.random.uniform(0.1, 10, num_segments-1))# 生成每段的幂指数exponents = np.random.uniform(-3, 3, num_segments)# 生成系数确保连续性coefficients = [1.0]for i in range(1, num_segments):prev_break = breakpoints[i-1] if i > 1 else 0.1coeff = coefficients[i-1] * (prev_break ** (exponents[i-1] - exponents[i]))coefficients.append(coeff)# 生成曲线数据num_points = np.random.randint(*self.num_points_range)x = np.logspace(-2, 2, num_points)  # 对数间隔的x值y = np.zeros_like(x)segment_start = 0for i, bp in enumerate(breakpoints + [x.max()]):segment_mask = (x >= (breakpoints[i-1] if i > 0 else x.min())) & (x <= bp)segment_x = x[segment_mask]y[segment_mask] = coefficients[i] * (segment_x ** exponents[i])# 添加噪声y_noisy = y * (1 + np.random.normal(0, self.noise_level, y.shape))return x, y_noisy, breakpoints, exponentsdef generate_synthetic_dataset(self, num_curves=10000):"""生成合成数据集"""curves = []labels = []for i in range(num_curves):# 随机选择分段数num_segments = np.random.randint(1, 5)# 生成曲线x, y, breakpoints, exponents = self.generate_piecewise_power_law(num_segments)# 转换为双对数坐标log_x = np.log10(x)log_y = np.log10(y)# 生成转折点标签turning_points = self.generate_turning_point_labels(log_x, log_y, breakpoints)curves.append((log_x, log_y))labels.append(turning_points)return curves, labelsdef generate_turning_point_labels(self, log_x, log_y, breakpoints):"""生成转折点标签"""labels = np.zeros(len(log_x))for bp in breakpoints:log_bp = np.log10(bp)# 找到最近的点的索引idx = np.argmin(np.abs(log_x - log_bp))# 设置转折点标签（考虑转折强度）strength = self.calculate_turning_strength(log_x, log_y, idx)labels[idx] = 2 if strength > 0.5 else 1  # 强转折或弱转折return labelsdef calculate_turning_strength(self, log_x, log_y, idx, window_size=5):"""计算转折点强度"""if idx < window_size or idx >= len(log_x) - window_size:return 0.0# 计算前后窗口的斜率差异prev_slope = np.polyfit(log_x[idx-window_size:idx], log_y[idx-window_size:idx], 1)[0]next_slope = np.polyfit(log_x[idx:idx+window_size], log_y[idx:idx+window_size], 1)[0]return abs(next_slope - prev_slope)

4.2 数据增强策略

为提高模型泛化能力，我们实施多种数据增强技术：

class CurveDataAugmentation:"""曲线数据增强"""def __init__(self):passdef add_noise(self, log_x, log_y, noise_type='gaussian', intensity=0.1):"""添加噪声"""if noise_type == 'gaussian':noise = np.random.normal(0, intensity, log_y.shape)return log_x, log_y + noiseelif noise_type == 'salt_pepper':# 椒盐噪声noise = np.zeros_like(log_y)salt_pepper_idx = np.random.choice(len(log_y), int(intensity * len(log_y)))noise[salt_pepper_idx] = np.random.choice([-intensity, intensity], len(salt_pepper_idx))return log_x, log_y + noisedef rescale_curve(self, log_x, log_y, scale_range=(0.8, 1.2)):"""曲线缩放"""scale_factor = np.random.uniform(*scale_range)return log_x, log_y * scale_factordef shift_curve(self, log_x, log_y, shift_range=(-0.1, 0.1)):"""曲线平移"""shift_amount = np.random.uniform(*shift_range)return log_x, log_y + shift_amountdef warp_curve(self, log_x, log_y, warp_intensity=0.1):"""曲线扭曲"""# 使用样条插值进行非线性扭曲t = np.linspace(0, 1, len(log_x))warp_points = np.random.normal(0, warp_intensity, 3)warp_spline = interpolate.CubicSpline([0, 0.5, 1], [0] + warp_points.tolist() + [0])warp = warp_spline(t)return log_x, log_y + warpdef apply_random_augmentation(self, log_x, log_y):"""应用随机增强组合"""augmented_x, augmented_y = log_x.copy(), log_y.copy()# 随机选择增强方法augmentations = [lambda x, y: self.add_noise(x, y, intensity=np.random.uniform(0.01, 0.1)),lambda x, y: self.rescale_curve(x, y, scale_range=(0.7, 1.3)),lambda x, y: self.shift_curve(x, y, shift_range=(-0.2, 0.2)),lambda x, y: self.warp_curve(x, y, warp_intensity=np.random.uniform(0.05, 0.2))]# 随机应用1-3种增强num_augmentations = np.random.randint(1, 4)selected_augmentations = np.random.choice(augmentations, num_augmentations, replace=False)for aug in selected_augmentations:augmented_x, augmented_y = aug(augmented_x, augmented_y)return augmented_x, augmented_y

4.3 数据加载器实现

from torch.utils.data import Dataset, DataLoader
import torchclass LogLogCurveDataset(Dataset):"""双对数坐标曲线数据集"""def __init__(self, curves, labels, augment=True, target_length=512):self.curves = curvesself.labels = labelsself.augment = augmentself.target_length = target_lengthself.augmentor = CurveDataAugmentation()def __len__(self):return len(self.curves)def __getitem__(self, idx):log_x, log_y = self.curves[idx]point_labels = self.labels[idx]# 数据增强if self.augment:log_x, log_y = self.augmentor.apply_random_augmentation(log_x, log_y)# 重采样到固定长度if len(log_x) != self.target_length:# 使用样条插值进行重采样t_original = np.linspace(0, 1, len(log_x))t_target = np.linspace(0, 1, self.target_length)# 对X和Y分别插值interp_x = interpolate.interp1d(t_original, log_x, kind='linear')interp_y = interpolate.interp1d(t_original, log_y, kind='linear')interp_labels = interpolate.interp1d(t_original, point_labels, kind='nearest')log_x_resampled = interp_x(t_target)log_y_resampled = interp_y(t_target)labels_resampled = interp_labels(t_target)else:log_x_resampled = log_xlog_y_resampled = log_ylabels_resampled = point_labels# 转换为张量curve_tensor = torch.tensor(np.stack([log_x_resampled, log_y_resampled]), dtype=torch.float32)labels_tensor = torch.tensor(labels_resampled, dtype=torch.long)# 生成曲线级别的置信度标签（基于转折点数量和强度）confidence_label = self.calculate_curve_confidence(labels_resampled)return {'curve': curve_tensor,'point_labels': labels_tensor,'curve_confidence': torch.tensor(confidence_label, dtype=torch.float32)}def calculate_curve_confidence(self, labels):"""计算曲线级别的置信度"""num_turning_points = np.sum(labels > 0)max_strength = np.max(labels) / 2.0  # 归一化到[0,1]# 综合考虑转折点数量和强度confidence = min(1.0, 0.3 * num_turning_points + 0.7 * max_strength)return confidence

5. 模型训练与优化

5.1 训练策略

我们采用分阶段训练策略，逐步优化模型性能：

class TurningPointTrainer:"""转折点检测模型训练器"""def __init__(self, model, train_loader, val_loader, device='cuda'):self.model = model.to(device)self.train_loader = train_loaderself.val_loader = val_loaderself.device = device# 优化器self.optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4, weight_decay=1e-4)# 学习率调度器self.scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(self.optimizer, T_0=10, T_mult=2)# 损失函数self.criterion = TurningPointLoss()# 训练记录self.train_losses = []self.val_losses = []self.best_val_loss = float('inf')def train_epoch(self, epoch):"""训练一个epoch"""self.model.train()total_loss = 0total_cls_loss = 0total_conf_loss = 0pbar = tqdm(self.train_loader, desc=f'Epoch {epoch}')for batch_idx, batch in enumerate(pbar):# 数据转移到设备curves = batch['curve'].to(self.device)point_labels = batch['point_labels'].to(self.device)curve_confidence = batch['curve_confidence'].to(self.device)# 前向传播predictions = self.model(curves)# 计算损失targets = {'point_labels': point_labels,'curve_confidence': curve_confidence}losses = self.criterion(predictions, targets)loss = losses['total_loss']# 反向传播self.optimizer.zero_grad()loss.backward()# 梯度裁剪torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)self.optimizer.step()# 记录损失total_loss += loss.item()total_cls_loss += losses['cls_loss_main'].item()total_conf_loss += losses['conf_loss'].item()# 更新进度条pbar.set_postfix({'Loss': f'{loss.item():.4f}','CLS': f'{losses["cls_loss_main"].item():.4f}','Conf': f'{losses["conf_loss"].item():.4f}'})avg_loss = total_loss / len(self.train_loader)avg_cls_loss = total_cls_loss / len(self.train_loader)avg_conf_loss = total_conf_loss / len(self.train_loader)self.train_losses.append(avg_loss)return avg_loss, avg_cls_loss, avg_conf_lossdef validate(self, epoch):"""验证模型"""self.model.eval()total_loss = 0total_accuracy = 0total_precision = 0total_recall = 0with torch.no_grad():for batch in self.val_loader:curves = batch['curve'].to(self.device)point_labels = batch['point_labels'].to(self.device)curve_confidence = batch['curve_confidence'].to(self.device)predictions = self.model(curves)targets = {'point_labels': point_labels,'curve_confidence': curve_confidence}losses = self.criterion(predictions, targets)total_loss += losses['total_loss'].item()# 计算评估指标accuracy, precision, recall = self.calculate_metrics(predictions['main_predictions'], point_labels)total_accuracy += accuracytotal_precision += precisiontotal_recall += recallavg_loss = total_loss / len(self.val_loader)avg_accuracy = total_accuracy / len(self.val_loader)avg_precision = total_precision / len(self.val_loader)avg_recall = total_recall / len(self.val_loader)self.val_losses.append(avg_loss)# 保存最佳模型if avg_loss < self.best_val_loss:self.best_val_loss = avg_losstorch.save({'epoch': epoch,'model_state_dict': self.model.state_dict(),'optimizer_state_dict': self.optimizer.state_dict(),'best_val_loss': self.best_val_loss,}, 'best_model.pth')return avg_loss, avg_accuracy, avg_precision, avg_recalldef calculate_metrics(self, predictions, targets):"""计算评估指标"""# 将预测转换为类别pred_classes = torch.argmax(predictions, dim=1)# 只考虑转折点（类别1和2）turning_point_mask = targets > 0if turning_point_mask.sum() == 0:return 0.0, 0.0, 0.0pred_turning = pred_classes[turning_point_mask]true_turning = targets[turning_point_mask]# 计算准确率accuracy = (pred_turning == true_turning).float().mean().item()# 计算精确率和召回率（针对转折点）true_positives = ((pred_turning > 0) & (true_turning > 0)).float().sum().item()false_positives = ((pred_turning > 0) & (true_turning == 0)).float().sum().item()false_negatives = ((pred_turning == 0) & (true_turning > 0)).float().sum().item()precision = true_positives / (true_positives + false_positives + 1e-8)recall = true_positives / (true_positives + false_negatives + 1e-8)return accuracy, precision, recalldef train(self, num_epochs):"""完整训练过程"""print("开始训练...")for epoch in range(num_epochs):# 训练阶段train_loss, train_cls, train_conf = self.train_epoch(epoch)# 验证阶段val_loss, val_acc, val_prec, val_rec = self.validate(epoch)# 更新学习率self.scheduler.step()# 打印训练信息print(f'Epoch {epoch+1}/{num_epochs}:')print(f'  训练损失: {train_loss:.4f} (分类: {train_cls:.4f}, 置信度: {train_conf:.4f})')print(f'  验证损失: {val_loss:.4f}, 准确率: {val_acc:.4f}')print(f'  精确率: {val_prec:.4f}, 召回率: {val_rec:.4f}')print(f'  学习率: {self.optimizer.param_groups[0]["lr"]:.6f}')

5.2 模型优化技巧

我们采用多种优化技巧提升模型性能：

class ModelOptimizer:"""模型优化工具类"""def __init__(self):passdef apply_gradual_unfreezing(self, model, current_epoch, total_epochs):"""渐进式解冻：逐步解冻网络层"""# 定义解冻计划unfreeze_schedule = {'backbone': 0.3,  # 30% epoch时解冻主干网络'attention': 0.6,  # 60% epoch时解冻注意力层'head': 0.8       # 80% epoch时解冻检测头}# 根据当前epoch决定解冻哪些层progress = current_epoch / total_epochsif progress < unfreeze_schedule['backbone']:# 只训练检测头for name, param in model.named_parameters():if 'detection_head' not in name and 'auxiliary_head' not in name:param.requires_grad = Falseelse:param.requires_grad = Trueelif progress < unfreeze_schedule['attention']:# 解冻注意力层for name, param in model.named_parameters():if 'attention' in name or 'detection_head' in name or 'auxiliary_head' in name:param.requires_grad = Trueelse:param.requires_grad = Falseelif progress < unfreeze_schedule['head']:# 解冻主干网络for param in model.parameters():param.requires_grad = Trueelse:# 全部解冻for param in model.parameters():param.requires_grad = Truedef apply_learning_rate_warmup(self, optimizer, current_step, warmup_steps, base_lr):"""学习率预热"""if current_step < warmup_steps:lr = base_lr * (current_step + 1) / warmup_stepsfor param_group in optimizer.param_groups:param_group['lr'] = lrdef apply_ema(self, model, ema_model, decay=0.999):"""指数移动平均"""model_params = dict(model.named_parameters())ema_params = dict(ema_model.named_parameters())for name in model_params:ema_params[name].data = (decay * ema_params[name].data + (1 - decay) * model_params[name].data)

6. 批量识别系统实现

6.1 批量处理框架

import os
import glob
from concurrent.futures import ThreadPoolExecutor
import pandas as pdclass BatchTurningPointDetector:"""批量转折点检测系统"""def __init__(self, model_path, device='cuda', batch_size=32, num_workers=4):self.device = deviceself.batch_size = batch_sizeself.num_workers = num_workers# 加载模型self.model = self.load_model(model_path)self.model.eval()# 结果存储self.results = []def load_model(self, model_path):"""加载训练好的模型"""model = LogLogTurningPointDetector()checkpoint = torch.load(model_path, map_location=self.device)model.load_state_dict(checkpoint['model_state_dict'])model.to(self.device)return modeldef process_single_curve(self, curve_data):"""处理单条曲线"""with torch.no_grad():# 预处理曲线数据curve_tensor = self.preprocess_curve(curve_data).unsqueeze(0).to(self.device)# 模型预测predictions = self.model(curve_tensor)# 后处理提取转折点turning_points = self.postprocess_predictions(predictions, curve_data)return turning_pointsdef preprocess_curve(self, curve_data):"""曲线预处理"""if isinstance(curve_data, tuple):log_x, log_y = curve_dataelse:# 假设输入是原始(x,y)数据，转换为双对数坐标x, y = curve_datalog_x = np.log10(x)log_y = np.log10(y)# 标准化和重采样preprocessor = CurvePreprocessor()curve_tensor = torch.tensor(np.stack([log_x, log_y]), dtype=torch.float32)processed = preprocessor(curve_tensor.unsqueeze(0)).squeeze(0)return processeddef postprocess_predictions(self, predictions, original_curve):"""预测结果后处理"""# 获取转折点概率point_probs = F.softmax(predictions['main_predictions'], dim=1)turning_probs = point_probs[0, 1:3, :].sum(dim=0)  # 弱转折和强转折概率之和# 使用非极大值抑制筛选转折点turning_points = self.non_maximum_suppression(turning_probs.cpu().numpy(), min_prob=0.3,  # 最小概率阈值min_distance=10  # 最小距离阈值)# 映射回原始坐标if isinstance(original_curve, tuple):log_x, log_y = original_curveelse:x, y = original_curvelog_x = np.log10(x)log_y = np.log10(y)# 将转折点索引映射回原始坐标original_turning_points = []for idx in turning_points:# 将重采样后的索引映射回原始索引original_idx = self.map_index_to_original(idx, len(turning_probs), len(log_x))original_turning_points.append({'index': original_idx,'x': 10**log_x[original_idx] if original_idx < len(log_x) else None,'y': 10**log_y[original_idx] if original_idx < len(log_y) else None,'log_x': log_x[original_idx] if original_idx < len(log_x) else None,'log_y': log_y[original_idx] if original_idx < len(log_y) else None,'probability': turning_probs[idx],'strength': 'strong' if point_probs[0, 2, idx] > point_probs[0, 1, idx] else 'weak'})return original_turning_pointsdef non_maximum_suppression(self, probabilities, min_prob=0.3, min_distance=10):"""非极大值抑制"""turning_points = []# 找到所有超过阈值的点candidate_indices = np.where(probabilities > min_prob)[0]while len(candidate_indices) > 0:# 选择概率最大的点max_idx = candidate_indices[np.argmax(probabilities[candidate_indices])]turning_points.append(max_idx)# 移除附近的其他点candidate_indices = candidate_indices[np.abs(candidate_indices - max_idx) > min_distance]return turning_pointsdef map_index_to_original(self, resampled_idx, resampled_len, original_len):"""将重采样索引映射回原始索引"""ratio = original_len / resampled_lenoriginal_idx = int(resampled_idx * ratio)return min(original_idx, original_len - 1)def process_batch(self, curve_list):"""批量处理曲线"""print(f"开始处理 {len(curve_list)} 条曲线...")# 使用多线程并行处理with ThreadPoolExecutor(max_workers=self.num_workers) as executor:results = list(executor.map(self.process_single_curve, curve_list))# 整理结果batch_results = []for i, turning_points in enumerate(results):batch_results.append({'curve_id': i,'num_turning_points': len(turning_points),'turning_points': turning_points,'processing_time': None  # 可以添加计时信息})self.results.extend(batch_results)return batch_resultsdef process_directory(self, data_dir, file_pattern="*.csv"):"""处理目录中的所有数据文件"""# 查找数据文件data_files = glob.glob(os.path.join(data_dir, file_pattern))print(f"找到 {len(data_files)} 个数据文件")all_curves = []for file_path in data_files:curves = self.load_curves_from_file(file_path)all_curves.extend(curves)# 分批处理batch_results = []for i in range(0, len(all_curves), self.batch_size):batch_curves = all_curves[i:i + self.batch_size]batch_result = self.process_batch(batch_curves)batch_results.extend(batch_result)print(f"已完成 {min(i + self.batch_size, len(all_curves))} / {len(all_curves)} 条曲线")return batch_resultsdef load_curves_from_file(self, file_path):"""从文件加载曲线数据"""# 支持多种格式的数据文件if file_path.endswith('.csv'):return self.load_csv_curves(file_path)elif file_path.endswith('.npy'):return self.load_numpy_curves(file_path)else:raise ValueError(f"不支持的文件格式: {file_path}")def load_csv_curves(self, file_path):"""从CSV文件加载曲线"""df = pd.read_csv(file_path)curves = []# 假设CSV包含多列，每列是一条曲线for column in df.columns:if column.startswith('curve_'):y_values = df[column].dropna().valuesx_values = np.arange(len(y_values))curves.append((x_values, y_values))return curvesdef save_results(self, output_path):"""保存识别结果"""if not self.results:print("没有结果可保存")return# 转换为DataFrame格式results_data = []for result in self.results:for tp in result['turning_points']:results_data.append({'curve_id': result['curve_id'],'x': tp['x'],'y': tp['y'],'log_x': tp['log_x'],'log_y': tp['log_y'],'probability': tp['probability'],'strength': tp['strength']})df = pd.DataFrame(results_data)df.to_csv(output_path, index=False)print(f"结果已保存到: {output_path}")

6.2 可视化与结果分析

class ResultVisualizer:"""结果可视化工具"""def __init__(self):self.fig_size = (12, 8)def plot_curve_with_turning_points(self, curve_data, turning_points, save_path=None, show_plot=True):"""绘制曲线并标记转折点"""fig, (ax1, ax2) = plt.subplots(1, 2, figsize=self.fig_size)if isinstance(curve_data, tuple):x, y = curve_datalog_x, log_y = np.log10(x), np.log10(y)else:log_x, log_y = curve_datax, y = 10**log_x, 10**log_y# 原始坐标图ax1.plot(x, y, 'b-', linewidth=1, label='曲线')for tp in turning_points:if tp['x'] is not None and tp['y'] is not None:color = 'red' if tp['strength'] == 'strong' else 'orange'ax1.plot(tp['x'], tp['y'], 'o', color=color, markersize=8, label=f'{tp["strength"]}转折点')ax1.set_xscale('log')ax1.set_yscale('log')ax1.set_xlabel('X (对数坐标)')ax1.set_ylabel('Y (对数坐标)')ax1.set_title('原始坐标下的转折点')ax1.legend()ax1.grid(True, which="both", ls="--")# 双对数坐标图ax2.plot(log_x, log_y, 'b-', linewidth=1, label='曲线')for tp in turning_points:if tp['log_x'] is not None and tp['log_y'] is not None:color = 'red' if tp['strength'] == 'strong' else 'orange'ax2.plot(tp['log_x'], tp['log_y'], 'o', color=color, markersize=8, label=f'{tp["strength"]}转折点')ax2.set_xlabel('log(X)')ax2.set_ylabel('log(Y)')ax2.set_title('双对数坐标下的转折点')ax2.legend()ax2.grid(True)plt.tight_layout()if save_path:plt.savefig(save_path, dpi=300, bbox_inches='tight')if show_plot:plt.show()else:plt.close()def plot_probability_heatmap(self, predictions, curve_data, save_path=None):"""绘制转折点概率热图"""fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10))log_x, log_y = curve_datapoint_probs = F.softmax(predictions['main_predictions'], dim=1)[0]# 曲线图ax1.plot(log_x, log_y, 'k-', linewidth=2)ax1.set_ylabel('log(Y)')ax1.set_title('双对数坐标曲线')ax1.grid(True)# 概率热图im = ax2.imshow(point_probs.detach().cpu().numpy(), aspect='auto', cmap='hot', interpolation='nearest',extent=[0, len(log_x), 0, 3])ax2.set_yticks([0.5, 1.5, 2.5])ax2.set_yticklabels(['非转折点', '弱转折点', '强转折点'])ax2.set_xlabel('数据点索引')ax2.set_title('转折点概率热图')plt.colorbar(im, ax=ax2, label='概率')plt.tight_layout()if save_path:plt.savefig(save_path, dpi=300, bbox_inches='tight')plt.show()def generate_statistical_report(self, results):"""生成统计报告"""if not results:print("没有结果可分析")return# 统计信息total_curves = len(results)total_turning_points = sum(r['num_turning_points'] for r in results)strong_points = sum(1 for r in results for tp in r['turning_points'] if tp['strength'] == 'strong')weak_points = total_turning_points - strong_points# 概率分布probabilities = [tp['probability'] for r in results for tp in r['turning_points']]print("=" * 50)print("转折点识别统计报告")print("=" * 50)print(f"分析曲线数量: {total_curves}")print(f"识别转折点总数: {total_turning_points}")print(f"强转折点: {strong_points} ({strong_points/total_turning_points*100:.1f}%)")print(f"弱转折点: {weak_points} ({weak_points/total_turning_points*100:.1f}%)")print(f"平均每条曲线转折点: {total_turning_points/total_curves:.2f}")print(f"转折点概率范围: {min(probabilities):.3f} - {max(probabilities):.3f}")print(f"平均概率: {np.mean(probabilities):.3f}")print("=" * 50)# 绘制概率分布直方图plt.figure(figsize=(10, 6))plt.hist(probabilities, bins=20, alpha=0.7, edgecolor='black')plt.xlabel('转折点概率')plt.ylabel('频数')plt.title('转折点概率分布')plt.grid(True, alpha=0.3)plt.show()

7. 性能评估与实验结果

7.1 评估指标设计

我们设计了全面的评估指标体系：

class TurningPointEvaluator:"""转折点识别性能评估器"""def __init__(self, tolerance_window=5):self.tolerance_window = tolerance_window  # 容忍窗口大小def evaluate_detection_performance(self, predictions, ground_truth):"""评估检测性能"""# 转换为转折点位置列表pred_positions = self.get_turning_point_positions(predictions)gt_positions = self.get_turning_point_positions(ground_truth)# 计算匹配true_positives, false_positives, false_negatives = self.match_points(pred_positions, gt_positions)# 计算指标precision = len(true_positives) / (len(true_positives) + len(false_positives) + 1e-8)recall = len(true_positives) / (len(true_positives) + len(false_negatives) + 1e-8)f1_score = 2 * precision * recall / (precision + recall + 1e-8)# 位置误差position_errors = self.calculate_position_errors(true_positives)return {'precision': precision,'recall': recall,'f1_score': f1_score,'true_positives': len(true_positives),'false_positives': len(false_positives),'false_negatives': len(false_negatives),'mean_position_error': np.mean(position_errors) if position_errors else 0,'std_position_error': np.std(position_errors) if position_errors else 0}def get_turning_point_positions(self, data):"""提取转折点位置"""positions = []if isinstance(data, dict) and 'turning_points' in data:for tp in data['turning_points']:positions.append(tp['index'])elif isinstance(data, np.ndarray):positions = np.where(data > 0)[0].tolist()elif isinstance(data, list):positions = [i for i, label in enumerate(data) if label > 0]return positionsdef match_points(self, pred_positions, gt_positions):"""匹配预测和真实转折点"""true_positives = []false_positives = []false_negatives = gt_positions.copy()for pred_pos in pred_positions:# 在容忍窗口内寻找匹配的真实点matched = Falsefor gt_pos in gt_positions:if abs(pred_pos - gt_pos) <= self.tolerance_window:true_positives.append((pred_pos, gt_pos))if gt_pos in false_negatives:false_negatives.remove(gt_pos)matched = Truebreakif not matched:false_positives.append(pred_pos)return true_positives, false_positives, false_negativesdef calculate_position_errors(self, true_positives):"""计算位置误差"""errors = [abs(pred - gt) for pred, gt in true_positives]return errorsdef evaluate_on_dataset(self, model, test_loader, device='cuda'):"""在整个测试集上评估模型"""model.eval()all_metrics = []with torch.no_grad():for batch in test_loader:curves = batch['curve'].to(device)ground_truth = batch['point_labels'].cpu().numpy()# 模型预测predictions = model(curves)pred_labels = torch.argmax(predictions['main_predictions'], dim=1).cpu().numpy()# 对每个样本计算指标for i in range(len(curves)):metrics = self.evaluate_detection_performance(pred_labels[i], ground_truth[i])all_metrics.append(metrics)# 计算平均指标avg_metrics = {}for key in all_metrics[0].keys():if key not in ['true_positives', 'false_positives', 'false_negatives']:avg_metrics[key] = np.mean([m[key] for m in all_metrics])else:avg_metrics[key] = np.sum([m[key] for m in all_metrics])return avg_metrics

7.2 对比实验设计

我们设计了与传统方法的对比实验：

class ComparativeExperiment:"""对比实验：深度学习 vs 传统方法"""def __init__(self):self.traditional_methods = {'derivative_based': DerivativeBasedDetector(),'curvature_based': CurvatureBasedDetector(),'statistical_based': StatisticalChangePointDetector()}def run_comparison(self, test_dataset):"""运行对比实验"""results = {}# 测试传统方法for method_name, detector in self.traditional_methods.items():print(f"测试传统方法: {method_name}")metrics = self.evaluate_method(detector, test_dataset)results[method_name] = metrics# 测试深度学习方法print("测试深度学习方法")# 这里需要加载训练好的深度学习模型# deep_metrics = self.evaluate_deep_learning_model(test_dataset)# results['deep_learning'] = deep_metricsreturn resultsdef evaluate_method(self, detector, test_dataset):"""评估单个方法"""all_metrics = []evaluator = TurningPointEvaluator()for curve_data, ground_truth in test_dataset:# 使用传统方法检测转折点turning_points = detector.detect(curve_data)# 评估性能metrics = evaluator.evaluate_detection_performance(turning_points, ground_truth)all_metrics.append(metrics)# 计算平均指标avg_metrics = self.average_metrics(all_metrics)return avg_metricsdef average_metrics(self, metrics_list):"""计算平均指标"""avg_metrics = {}for key in metrics_list[0].keys():if key not in ['true_positives', 'false_positives', 'false_negatives']:avg_metrics[key] = np.mean([m[key] for m in metrics_list])else:avg_metrics[key] = np.sum([m[key] for m in metrics_list])return avg_metricsdef plot_comparison_results(self, results):"""绘制对比结果"""methods = list(results.keys())metrics = ['precision', 'recall', 'f1_score']fig, axes = plt.subplots(1, 3, figsize=(15, 5))for i, metric in enumerate(metrics):values = [results[method][metric] for method in methods]axes[i].bar(methods, values, color=['skyblue', 'lightgreen', 'lightcoral', 'gold'])axes[i].set_title(f'{metric.upper()} 对比')axes[i].set_ylabel(metric)axes[i].tick_params(axis='x', rotation=45)# 在柱子上添加数值for j, v in enumerate(values):axes[i].text(j, v + 0.01, f'{v:.3f}', ha='center', va='bottom')plt.tight_layout()plt.show()# 传统方法实现示例
class DerivativeBasedDetector:"""基于导数的方法"""def detect(self, curve_data):# 实现基于导数的转折点检测passclass CurvatureBasedDetector:"""基于曲率的方法"""def detect(self, curve_data):# 实现基于曲率的转折点检测passclass StatisticalChangePointDetector:"""基于统计变点检测的方法"""def detect(self, curve_data):# 实现基于统计的变点检测pass

8. 应用案例与系统部署

8.1 实际应用案例

我们将该系统应用于几个实际场景：

class RealWorldApplications:"""实际应用案例展示"""def analyze_material_stress_strain(self, data_file):"""分析材料应力-应变曲线"""print("分析材料应力-应变曲线...")# 加载数据stress_strain_data = self.load_stress_strain_data(data_file)# 检测转折点（弹性到塑性转变）detector = BatchTurningPointDetector('trained_model.pth')results = detector.process_batch([stress_strain_data])# 分析结果for turning_point in results[0]['turning_points']:if turning_point['strength'] == 'strong':print(f"发现材料屈服点: 应力 = {turning_point['y']:.2f} MPa, "f"应变 = {turning_point['x']:.4f}")return resultsdef analyze_seismic_spectra(self, seismic_data):"""分析地震波谱"""print("分析地震波谱曲线...")# 地震波谱通常显示不同频率成分的转折detector = BatchTurningPointDetector('trained_model.pth')results = detector.process_batch(seismic_data)# 识别不同地层界面对应的转折点for i, result in enumerate(results):strong_points = [tp for tp in result['turning_points'] if tp['strength'] == 'strong']print(f"地震记录 {i}: 发现 {len(strong_points)} 个主要地层界面")return resultsdef analyze_economic_data(self, economic_series):"""分析经济数据序列"""print("分析经济规模效应曲线...")# 经济数据中的转折点可能表示规模收益的变化detector = BatchTurningPointDetector('trained_model.pth')results = detector.process_batch(economic_series)for i, result in enumerate(results):turning_points = result['turning_points']if turning_points:print(f"经济序列 {i}: 发现 {len(turning_points)} 个规模效应转变点")return resultsdef load_stress_strain_data(self, file_path):"""加载应力-应变数据"""# 实际实现中会从文件读取真实数据# 这里使用模拟数据strain = np.linspace(0, 0.2, 200)stress = 200 * strain * (1 - 0.5 * strain)  # 模拟弹塑性行为return (strain, stress)

8.2 系统部署方案

class DeploymentSystem:"""系统部署方案"""def create_web_service(self, model_path, host='localhost', port=5000):"""创建Web服务API"""from flask import Flask, request, jsonifyimport base64import ioapp = Flask(__name__)detector = BatchTurningPointDetector(model_path)@app.route('/api/detect_turning_points', methods=['POST'])def detect_turning_points():# 接收JSON格式的曲线数据data = request.jsoncurves = data.get('curves', [])# 处理曲线results = detector.process_batch(curves)return jsonify({'success': True,'results': results,'message': f'成功处理 {len(curves)} 条曲线'})@app.route('/api/upload_file', methods=['POST'])def upload_file():# 处理文件上传file = request.files['file']file_type = file.filename.split('.')[-1]# 根据文件类型处理if file_type == 'csv':curves = self.parse_csv_file(file)elif file_type in ['txt', 'dat']:curves = self.parse_text_file(file)else:return jsonify({'success': False, 'error': '不支持的文件格式'})results = detector.process_batch(curves)return jsonify({'success': True,'filename': file.filename,'results': results})print(f"启动Web服务: http://{host}:{port}")app.run(host=host, port=port, debug=False)def create_desktop_application(self):"""创建桌面应用程序"""try:import tkinter as tkfrom tkinter import filedialog, messageboximport matplotlib.pyplot as pltfrom matplotlib.backends.backend_tkagg import FigureCanvasTkAggclass TurningPointApp:def __init__(self, root):self.root = rootself.root.title("双对数坐标曲线转折点识别系统")self.root.geometry("1200x800")# 加载模型self.detector = BatchTurningPointDetector('trained_model.pth')self.visualizer = ResultVisualizer()self.create_widgets()def create_widgets(self):# 创建界面组件self.create_menu()self.create_main_frame()def create_menu(self):menubar = tk.Menu(self.root)file_menu = tk.Menu(menubar, tearoff=0)file_menu.add_command(label="打开文件", command=self.open_file)file_menu.add_separator()file_menu.add_command(label="退出", command=self.root.quit)menubar.add_cascade(label="文件", menu=file_menu)self.root.config(menu=menubar)def create_main_frame(self):# 主界面框架main_frame = tk.Frame(self.root)main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)# 控制面板control_frame = tk.Frame(main_frame)control_frame.pack(fill=tk.X, pady=5)tk.Button(control_frame, text="批量处理目录", command=self.batch_process).pack(side=tk.LEFT, padx=5)# 结果显示区域self.result_text = tk.Text(main_frame, height=15)self.result_text.pack(fill=tk.BOTH, expand=True, pady=5)def open_file(self):filename = filedialog.askopenfilename(title="选择数据文件",filetypes=[("CSV文件", "*.csv"), ("文本文件", "*.txt"), ("所有文件", "*.*")])if filename:self.process_single_file(filename)def process_single_file(self, filename):try:curves = self.detector.load_curves_from_file(filename)results = self.detector.process_batch(curves)# 显示结果self.display_results(results, filename)# 可视化第一条曲线if curves:self.visualizer.plot_curve_with_turning_points(curves[0], results[0]['turning_points'])except Exception as e:messagebox.showerror("错误", f"处理文件时出错: {str(e)}")def batch_process(self):directory = filedialog.askdirectory(title="选择数据目录")if directory:results = self.detector.process_directory(directory)self.detector.save_results("batch_results.csv")messagebox.showinfo("完成", f"批量处理完成，共处理 {len(results)} 条曲线")def display_results(self, results, filename):self.result_text.delete(1.0, tk.END)self.result_text.insert(tk.END, f"文件: {filename}\n")self.result_text.insert(tk.END, "="*50 + "\n")for i, result in enumerate(results):self.result_text.insert(tk.END, f"曲线 {i}:\n")self.result_text.insert(tk.END, f"  转折点数量: {result['num_turning_points']}\n")for j, tp in enumerate(result['turning_points']):self.result_text.insert(tk.END, f"  转折点 {j}: 位置=({tp['x']:.4f}, {tp['y']:.4f}), "f"概率={tp['probability']:.3f}, 强度={tp['strength']}\n")self.result_text.insert(tk.END, "\n")root = tk.Tk()app = TurningPointApp(root)root.mainloop()except ImportError:print("GUI库不可用，无法启动桌面应用")