C1.py

import numpy as np
import pandas as pd
import onnxruntime as ort
import os
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns
import glob

def load_raw_sensor_data(data_path):
    """
    加载UCI HAR数据集的原始传感器数据
    
    参数:
    data_path: 数据集根目录路径或直接是Inertial Signals目录路径
    
    返回:
    X_raw: 原始传感器数据
    y_test: 测试标签数据
    """
    # 确定Inertial Signals目录路径
    if os.path.basename(data_path) == 'Inertial Signals':
        inertial_signals_path = data_path
        # 尝试推断测试标签路径
        y_test_path = os.path.join(os.path.dirname(data_path), 'y_test.txt')
    else:
        # 假设提供的是数据集根目录
        inertial_signals_path = os.path.join(data_path, 'test', 'Inertial Signals')
        y_test_path = os.path.join(data_path, 'test', 'y_test.txt')
    
    # 检查路径是否存在
    if not os.path.exists(inertial_signals_path):
        raise FileNotFoundError(f"找不到原始传感器数据目录: {inertial_signals_path}")
    
    # 加载测试标签
    if os.path.exists(y_test_path):
        y_test = pd.read_csv(y_test_path, header=None)[0].values
        print(f"加载了 {len(y_test)} 个测试标签")
    else:
        print(f"警告: 找不到测试标签文件 {y_test_path}")
        print("将无法计算准确率，仅执行预测")
        y_test = None
    
    # 获取所有传感器数据文件
    sensor_files = glob.glob(os.path.join(inertial_signals_path, '*.txt'))
    if not sensor_files:
        raise FileNotFoundError(f"在 {inertial_signals_path} 中没有找到传感器数据文件")
    
    print(f"找到 {len(sensor_files)} 个传感器数据文件:")
    for f in sensor_files:
        print(f"  - {os.path.basename(f)}")
    
    # 加载每个传感器文件
    sensor_data = []
    for file_path in sorted(sensor_files):  # 排序以确保顺序一致
        data = pd.read_csv(file_path, delim_whitespace=True, header=None).values
        sensor_data.append(data)
        print(f"加载 {os.path.basename(file_path)}: 形状 {data.shape}")
    
    # 创建不同格式的数据组合
    # 1. 堆叠格式: [samples, timesteps, sensors]
    X_stacked = np.dstack(sensor_data)
    print(f"堆叠格式数据形状: {X_stacked.shape}")
    
    # 2. 连接格式: [samples, timesteps*sensors]
    X_concat = np.concatenate(sensor_data, axis=1)
    print(f"连接格式数据形状: {X_concat.shape}")
    
    # 3. 转置堆叠格式: [samples, sensors, timesteps]
    X_transposed = X_stacked.transpose(0, 2, 1)
    print(f"转置堆叠格式数据形状: {X_transposed.shape}")
    
    return {
        'stacked': X_stacked.astype(np.float32),  # [samples, timesteps, sensors]
        'concatenated': X_concat.astype(np.float32),  # [samples, timesteps*sensors]
        'transposed': X_transposed.astype(np.float32)  # [samples, sensors, timesteps]
    }, y_test

def evaluate_onnx_model(model_path, X_data, y_test, data_format='all'):
    """
    评估ONNX模型性能
    
    参数:
    model_path: ONNX模型文件路径
    X_data: 测试数据 (可以是字典包含多种格式或单个数组)
    y_test: 测试标签数据
    data_format: 要使用的数据格式 ('all', 'stacked', 'concatenated', 'transposed')
    
    返回:
    best_result: 包含最佳结果的字典
    """
    # 检查模型文件是否存在
    if not os.path.exists(model_path):
        raise FileNotFoundError(f"找不到模型文件: {model_path}")
    
    # 创建ONNX运行时会话
    print(f"加载ONNX模型: {model_path}")
    session = ort.InferenceSession(model_path)
    
    # 获取输入和输出名称
    input_name = session.get_inputs()[0].name
    print(f"模型输入名称: {input_name}")
    
    # 获取模型输入形状
    input_shape = session.get_inputs()[0].shape
    print(f"模型输入形状: {input_shape}")
    
    # 确定要尝试的数据格式
    formats_to_try = []
    
    if isinstance(X_data, dict):
        if data_format == 'all':
            formats_to_try = list(X_data.keys())
        elif data_format in X_data:
            formats_to_try = [data_format]
        else:
            print(f"警告: 未找到指定的数据格式 '{data_format}'，将尝试所有可用格式")
            formats_to_try = list(X_data.keys())
    else:
        # 如果X_data不是字典，直接使用它
        formats_to_try = ['single']
        X_data = {'single': X_data}
    
    # 存储最佳结果
    best_result = {
        'accuracy': 0,
        'report': None,
        'conf_matrix': None,
        'y_pred': None,
        'format': None
    }
    
    # 尝试每种数据格式
    for fmt in formats_to_try:
        print(f"\n尝试数据格式: {fmt}")
        X_test = X_data[fmt]
        
        # 根据模型输入形状调整数据
        original_shape = X_test.shape
        modified_shape = original_shape
        
        # 检查模型是否期望3D输入 (batch_size, seq_len, features)
        try:
            if len(input_shape) == 3:
                # 如果输入形状是 [batch_size, seq_len, features]
                if fmt == 'stacked':
                    # 数据已经是 [samples, timesteps, sensors] 格式
                    pass
                elif fmt == 'transposed':
                    # 数据是 [samples, sensors, timesteps] 格式，可能需要转置
                    if input_shape[1] != -1 and input_shape[1] == X_test.shape[2]:
                        # 需要转置 [samples, sensors, timesteps] -> [samples, timesteps, sensors]
                        X_test = X_test.transpose(0, 2, 1)
                        modified_shape = X_test.shape
                elif fmt == 'concatenated' or fmt == 'single':
                    # 数据是2D的，需要变成3D
                    if len(original_shape) == 2:
                        if input_shape[1] != -1 and input_shape[2] != -1:
                            # 尝试重塑为指定的seq_len和features
                            if original_shape[1] == input_shape[1] * input_shape[2]:
                                X_test = X_test.reshape(original_shape[0], input_shape[1], input_shape[2])
                                modified_shape = X_test.shape
            
            # 如果模型期望2D输入，但数据是3D的
            elif len(input_shape) == 2 and len(original_shape) == 3:
                # 展平数据
                X_test = X_test.reshape(original_shape[0], -1)
                modified_shape = X_test.shape
            
            print(f"原始形状: {original_shape} -> 修改后形状: {modified_shape}")
        
        except Exception as e:
            print(f"调整数据形状时出错: {e}")
            print("将使用原始形状")
        
        # 运行预测
        try:
            print("运行模型预测...")
            predictions = session.run(None, {input_name: X_test})[0]
            
            # 处理预测结果
            if len(predictions.shape) > 1 and predictions.shape[1] > 1:
                # 如果输出是类别概率，取最大概率的类别
                y_pred = np.argmax(predictions, axis=1)
                
                # 如果标签从1开始而预测从0开始，需要调整
                if y_test is not None and np.min(y_test) == 1 and np.min(y_pred) == 0:
                    y_pred = y_pred + 1
            else:
                # 如果输出是直接的类别标签
                y_pred = predictions.ravel().astype(int)
                
                # 调整标签范围（如果需要）
                if y_test is not None and np.min(y_test) == 1 and np.min(y_pred) == 0:
                    y_pred = y_pred + 1
            
            # 如果有标签数据，计算准确率
            if y_test is not None:
                accuracy = accuracy_score(y_test, y_pred)
                print(f"格式 '{fmt}' 的准确率: {accuracy:.4f}")
                
                # 活动标签
                activity_labels = {
                    1: 'WALKING',
                    2: 'WALKING_UPSTAIRS',
                    3: 'WALKING_DOWNSTAIRS',
                    4: 'SITTING',
                    5: 'STANDING',
                    6: 'LAYING'
                }
                
                # 生成分类报告
                class_names = [activity_labels[i] for i in sorted(activity_labels.keys())]
                report = classification_report(y_test, y_pred, target_names=class_names)
                
                # 计算混淆矩阵
                conf_matrix = confusion_matrix(y_test, y_pred)
                
                # 更新最佳结果
                if accuracy > best_result['accuracy']:
                    best_result['accuracy'] = accuracy
                    best_result['report'] = report
                    best_result['conf_matrix'] = conf_matrix
                    best_result['y_pred'] = y_pred
                    best_result['format'] = fmt
                    best_result['class_names'] = class_names
            else:
                print(f"成功生成格式 '{fmt}' 的预测结果（无法计算准确率）")
                # 由于没有标签数据，只保存预测结果
                best_result['y_pred'] = y_pred
                best_result['format'] = fmt
        
        except Exception as e:
            print(f"使用格式 '{fmt}' 评估模型时出错: {e}")
            print("尝试下一个数据格式")
    
    # 如果找到了最佳结果
    if best_result['accuracy'] > 0:
        print(f"\n最佳结果来自格式: {best_result['format']}")
        print(f"最佳准确率: {best_result['accuracy']:.4f}")
        print("\n分类报告:")
        print(best_result['report'])
    elif best_result['y_pred'] is not None:
        print(f"\n生成了预测结果，使用的格式: {best_result['format']}")
    else:
        print("\n所有数据格式都未能成功评估模型")
    
    return best_result

def plot_confusion_matrix(conf_matrix, class_names):
    """
    绘制混淆矩阵
    
    参数:
    conf_matrix: 混淆矩阵
    class_names: 类别名称列表
    """
    plt.figure(figsize=(10, 8))
    sns.heatmap(conf_matrix, annot=True, fmt='d', cmap='Blues',
                xticklabels=class_names, yticklabels=class_names)
    plt.title('Confusion Matrix')
    plt.ylabel('Actual Activity')
    plt.xlabel('Predicted Activity')
    plt.tight_layout()
    plt.savefig('confusion_matrix.png')
    plt.close()
    print("混淆矩阵已保存为 confusion_matrix.png")

def plot_activity_accuracy(y_test, y_pred, class_names):
    """
    绘制各活动的准确率
    
    参数:
    y_test: 真实标签
    y_pred: 预测标签
    class_names: 类别名称列表
    """
    # 计算每个活动的准确率
    activity_accuracy = {}
    for i, class_name in enumerate(class_names, 1):
        mask = (y_test == i)
        if np.sum(mask) > 0:
            acc = accuracy_score(y_test[mask], y_pred[mask])
            activity_accuracy[class_name] = acc
    
    # 绘制准确率柱状图
    plt.figure(figsize=(10, 6))
    activities = list(activity_accuracy.keys())
    accuracies = list(activity_accuracy.values())
    
    # 按准确率排序
    sorted_indices = np.argsort(accuracies)
    sorted_activities = [activities[i] for i in sorted_indices]
    sorted_accuracies = [accuracies[i] for i in sorted_indices]
    
    plt.barh(sorted_activities, sorted_accuracies, color='skyblue')
    
    # 在条形上添加文本标签
    for i, v in enumerate(sorted_accuracies):
        plt.text(v + 0.01, i, f'{v:.2f}', va='center')
    
    plt.title('Accuracy by Activity')
    plt.xlabel('Accuracy')
    plt.ylabel('Activity')
    plt.xlim(0, 1.1)
    plt.grid(axis='x', linestyle='--', alpha=0.6)
    plt.tight_layout()
    plt.savefig('activity_accuracy.png')
    plt.close()
    print("各活动准确率已保存为 activity_accuracy.png")

def main():
    """
    主函数
    """
    # 配置参数
    model_path = "lstm_model_opt.onnx"  # ONNX模型路径
    data_path = "./datasets/test/Inertial Signals"  # 传感器数据路径
    
    # 检查是否存在命令行参数
    import sys
    if len(sys.argv) > 1:
        model_path = sys.argv[1]
    if len(sys.argv) > 2:
        data_path = sys.argv[2]
    
    # 检查模型文件
    if not os.path.exists(model_path):
        print(f"错误: 找不到模型文件 {model_path}")
        model_path = input("请输入ONNX模型的路径: ")
        if not os.path.exists(model_path):
            print(f"错误: 找不到模型文件 {model_path}")
            return
    
    # 检查数据路径
    inertial_signals_path = data_path
    if not os.path.exists(inertial_signals_path):
        print(f"警告: 找不到传感器数据目录 {inertial_signals_path}")
        # 尝试在当前目录下查找
        potential_paths = [
            "./test/Inertial Signals",
            "./UCI HAR Dataset/test/Inertial Signals",
            "./data/test/Inertial Signals"
        ]
        
        for path in potential_paths:
            if os.path.exists(path):
                inertial_signals_path = path
                print(f"找到传感器数据目录: {inertial_signals_path}")
                break
        
        if not os.path.exists(inertial_signals_path):
            inertial_signals_path = input("请输入传感器数据目录的路径: ")
            if not os.path.exists(inertial_signals_path):
                print(f"错误: 找不到传感器数据目录 {inertial_signals_path}")
                return
    
    try:
        # 加载原始传感器数据
        print("加载原始传感器数据...")
        X_data, y_test = load_raw_sensor_data(inertial_signals_path)
        
        # 评估模型
        print("\n评估模型...")
        best_result = evaluate_onnx_model(model_path, X_data, y_test)
        
        if best_result['y_pred'] is not None:
            # 保存预测结果
            results_df = pd.DataFrame({
                'Predicted': best_result['y_pred']
            })
            
            if y_test is not None:
                results_df['True'] = y_test
            
            results_df.to_csv('prediction_results.csv', index=False)
            print("预测结果已保存到 prediction_results.csv")
            
            # 如果有足够的评估数据，绘制结果
            if y_test is not None and best_result['conf_matrix'] is not None:
                # 绘制混淆矩阵
                plot_confusion_matrix(best_result['conf_matrix'], best_result['class_names'])
                
                # 绘制各活动的准确率
                plot_activity_accuracy(y_test, best_result['y_pred'], best_result['class_names'])
                
                print("\n评估完成！")
                print(f"最佳数据格式: {best_result['format']}")
                print(f"最佳准确率: {best_result['accuracy']:.4f}")
        
    except Exception as e:
        print(f"程序执行过程中出错: {e}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    main()

if __name__ == "__main__":
    main()