{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":97984,"databundleVersionId":14096757,"sourceType":"competition"}],"dockerImageVersionId":31192,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-11-26T12:03:18.765562Z","iopub.execute_input":"2025-11-26T12:03:18.765919Z","iopub.status.idle":"2025-11-26T12:03:30.069493Z","shell.execute_reply.started":"2025-11-26T12:03:18.765892Z","shell.execute_reply":"2025-11-26T12:03:30.068482Z"},"collapsed":true,"jupyter":{"outputs_hidden":true}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import cv2\nimport numpy as np\nimport pandas as pd\nimport pyarrow.parquet as pq\nimport os\nfrom scipy.signal import correlate\nfrom scipy.ndimage import shift\nfrom tensorflow.keras.models import Model\nfrom tensorflow.keras.layers import Input, Conv2D, MaxPooling2D, Conv2DTranspose\n\n# ==================== 数据路径配置（标注修改点）====================\nTRAIN_CSV = '/kaggle/input/physionet-ecg-image-digitization/train.csv'\nTEST_CSV = '/kaggle/input/physionet-ecg-image-digitization/test.csv'\nSAMPLE_PARQUET = '/kaggle/input/physionet-ecg-image-digitization/sample_submission.parquet'\n# ===================================================================\n\n# 数据加载函数\ndef load_data():\n    # 加载训练数据\n    if os.path.exists(TRAIN_CSV):\n        train_df = pd.read_csv(TRAIN_CSV)\n    else:\n        raise FileNotFoundError(f\"训练数据文件不存在: {TRAIN_CSV}\")\n    \n    # 加载测试数据\n    if os.path.exists(TEST_CSV):\n        test_df = pd.read_csv(TEST_CSV)\n    else:\n        raise FileNotFoundError(f\"测试数据文件不存在: {TEST_CSV}\")\n    \n    # 加载示例提交文件\n    if os.path.exists(SAMPLE_PARQUET):\n        sample_sub = pq.read_table(SAMPLE_PARQUET).to_pandas()\n    else:\n        sample_sub = pd.DataFrame({'id': [], 'value': []})\n    \n    return train_df, test_df, sample_sub\n\n# ECG图像预处理\ndef preprocess_ecg_image(image_path, target_size=(1000, 400)):\n    \"\"\"预处理ECG图像：灰度化、缩放、归一化\"\"\"\n    if not os.path.exists(image_path):\n        print(f\"警告：图像文件不存在 {image_path}\")\n        return np.zeros((target_size[1], target_size[0], 1))\n    \n    img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE)\n    img = cv2.resize(img, target_size)\n    img = (img - np.mean(img)) / (np.std(img) + 1e-8)  # 标准化\n    return img[..., np.newaxis]\n\n# 构建轻量级U-Net模型\ndef build_ecg_extractor(input_shape=(400, 1000, 1)):\n    \"\"\"构建用于ECG信号提取的轻量级U-Net模型\"\"\"\n    inputs = Input(shape=input_shape)\n    \n    # 编码器\n    x = Conv2D(16, (3, 3), activation='relu', padding='same')(inputs)\n    x = MaxPooling2D((2, 2))(x)\n    \n    x = Conv2D(32, (3, 3), activation='relu', padding='same')(x)\n    x = MaxPooling2D((2, 2))(x)\n    \n    # 解码器\n    x = Conv2DTranspose(32, (2, 2), strides=(2, 2), padding='same')(x)\n    x = Conv2D(16, (3, 3), activation='relu', padding='same')(x)\n    \n    x = Conv2DTranspose(16, (2, 2), strides=(2, 2), padding='same')(x)\n    outputs = Conv2D(1, (1, 1), activation='linear')(x)\n    \n    model = Model(inputs, outputs)\n    model.compile(optimizer='adam', loss='mse')\n    return model\n\n# 竞赛标准SNR计算\ndef calculate_corrected_snr(pred_signal, true_signal, max_time_shift=0.2, sampling_rate=500):\n    \"\"\"\n    计算竞赛标准的修正SNR\n    :param pred_signal: 预测信号\n    :param true_signal: 真实信号\n    :param max_time_shift: 最大时间偏移（秒）\n    :param sampling_rate: 采样率（Hz）\n    \"\"\"\n    # 1. 时间对齐\n    max_shift_samples = int(max_time_shift * sampling_rate)\n    corr = correlate(pred_signal, true_signal, mode='full')\n    shift_idx = np.argmax(corr) - (len(true_signal) - 1)\n    shift_idx = np.clip(shift_idx, -max_shift_samples, max_shift_samples)\n    \n    pred_shifted = shift(pred_signal, shift_idx)\n    \n    # 2. 垂直对齐\n    pred_shifted -= np.mean(pred_shifted - true_signal)\n    \n    # 3. 计算SNR\n    signal_power = np.sum(true_signal ** 2)\n    noise_power = np.sum((pred_shifted - true_signal) ** 2)\n    \n    if noise_power == 0:\n        return np.inf\n    return 10 * np.log10(signal_power / noise_power)\n\n# 主流程\ndef main():\n    # 加载数据\n    print(\"加载数据...\")\n    train_df, test_df, sample_sub = load_data()\n    \n    # 初始化模型\n    print(\"构建模型...\")\n    model = build_ecg_extractor()\n    \n    # 训练模型（使用前200个样本快速训练）\n    if not train_df.empty and 'image_path' in train_df.columns and 'signal_path' in train_df.columns:\n        print(\"训练模型...\")\n        train_samples = train_df.head(200)\n        \n        # 准备训练数据\n        X_train = np.array([preprocess_ecg_image(path) for path in train_samples['image_path']])\n        y_train = np.array([np.load(path).reshape(400, 1000, 1) for path in train_samples['signal_path']])\n        \n        # 训练模型\n        model.fit(X_train, y_train, epochs=10, batch_size=8, verbose=1)\n    \n    # 生成测试集预测\n    print(\"生成预测结果...\")\n    submission = []\n    \n    for _, row in test_df.iterrows():\n        # 处理每个测试样本\n        if 'image_path' in row:\n            img = preprocess_ecg_image(row['image_path'])\n            pred_signal = model.predict(img[np.newaxis, ...], verbose=0).flatten()\n        else:\n            pred_signal = np.zeros(400000)\n        \n        # 获取ID\n        current_id = row['id'] if 'id' in row else f\"unknown_{_}\"\n        \n        # 计算SNR（测试集无真实信号时使用示例值，实际竞赛需根据模型预测）\n        # 注：此处为演示，实际需替换为真实信号计算\n        snr_value = np.random.uniform(0, 10)  # 竞赛中需替换为真实计算\n        \n        submission.append({'id': current_id, 'value': snr_value})\n    \n    # 保存提交文件\n    submission_df = pd.DataFrame(submission)\n    submission_df.to_csv('submission.csv', index=False)\n    print(\"提交文件已保存为 submission.csv\")\n    print(submission_df.head())\n\nif __name__ == \"__main__\":\n    main()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-26T12:03:30.07114Z","iopub.execute_input":"2025-11-26T12:03:30.071504Z","iopub.status.idle":"2025-11-26T12:03:30.213787Z","shell.execute_reply.started":"2025-11-26T12:03:30.071472Z","shell.execute_reply":"2025-11-26T12:03:30.212654Z"}},"outputs":[],"execution_count":null}]}