人工智能在医疗中的应用-基于深度学习的心电图异常检测全解析

# 代码示例：加载心电图数据集
import wfdb

# 下载MIT-BIH心律失常数据库
record = wfdb.rdrecord('mitdb/100', channels=[0])

# 打印数据信息
print(record.__dict__)

# 代码示例：心电图数据预处理
import numpy as np
from scipy import signal

# 应用带通滤波器
fs = record.fs  # 采样率
lowcut = 0.5
highcut = 40.0
nyquist = 0.5 * fs
low = lowcut / nyquist
high = highcut / nyquist
b, a = signal.butter(1, [low, high], btype='band')
filtered_ecg = signal.filtfilt(b, a, record.p_signal[:, 0])

# 数据标准化
normalized_ecg = (filtered_ecg - np.mean(filtered_ecg)) / np.std(filtered_ecg)

# 代码示例：基于自编码器的异常检测模型
import tensorflow as tf
from tensorflow.keras.layers import Input, Dense
from tensorflow.keras.models import Model

# 构建自编码器模型
input_shape = (filtered_ecg.shape[0],)
input_ecg = Input(shape=input_shape)
encoded = Dense(128, activation='relu')(input_ecg)
decoded = Dense(input_shape[0], activation='sigmoid')(encoded)

autoencoder = Model(input_ecg, decoded)
autoencoder.compile(optimizer='adam', loss='mse')

# 模型训练
autoencoder.fit(normalized_ecg, normalized_ecg, epochs=50, batch_size=128, shuffle=True, validation_split=0.2)

# 使用训练好的模型进行异常检测
reconstructed_ecg = autoencoder.predict(normalized_ecg)
mse = np.mean(np.power(normalized_ecg - reconstructed_ecg, 2), axis=1)
threshold = np.mean(mse) + np.std(mse) * 2  # 根据重构误差设定异常阈值

# 标记异常数据点
anomalies = np.where(mse > threshold)[0]
print("Detected anomalies:", anomalies)

# 代码示例：模型评估
from sklearn.metrics import confusion_matrix, classification_report

# 定义正常和异常数据标签
labels = np.zeros_like(mse)
labels[mse > threshold] = 1

# 计算混淆矩阵
cm = confusion_matrix(y_true, labels)
print("Confusion Matrix:")
print(cm)

# 计算分类报告
print("Classification Report:")
print(classification_report(y_true, labels))

# 代码示例：模型部署
# （这部分代码可以是一个基本的部署模板，例如使用Flask搭建一个简单的Web应用）
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/predict', methods=['POST'])
def predict():
    data = request.json['data']  # 获取POST请求中的数据
    # 在此处应用预处理和模型预测逻辑
    return jsonify({'result': 'normal' if is_normal(data) else 'anomaly'})

if __name__ == '__main__':
    app.run(debug=True)

# 代码示例：实时心电图异常检测
def real_time_ecg_anomaly_detection(new_ecg_data):
    # 对新的心电图数据进行预处理
    # ...

    # 使用训练好的模型进行预测
    prediction = autoencoder.predict(new_ecg_data)

    # 计算重构误差
    mse_new = np.mean(np.power(new_ecg_data - prediction, 2), axis=1)

    # 判断是否异常
    if np.mean(mse_new) > threshold:
        return "Anomaly detected"
    else:
        return "Normal"

# 调用实时监测函数示例
new_ecg_data = load_new_ecg_data()  # 加载新的心电图数据
result = real_time_ecg_anomaly_detection(new_ecg_data)
print("Real-time anomaly detection result:", result)

# 代码示例：模型集成到医疗信息系统
from database import ECGRecord, save_anomaly_detection_result

def process_ecg_record(record_id):
    # 从数据库加载心电图记录
    ecg_data = ECGRecord.load(record_id)

    # 调用异常检测函数
    result = real_time_ecg_anomaly_detection(ecg_data)

    # 将检测结果保存到数据库
    save_anomaly_detection_result(record_id, result)

# 代码示例：实时心电图异常检测
def real_time_ecg_anomaly_detection(new_ecg_data):
    # 对新的心电图数据进行预处理
    # ...

    # 使用训练好的模型进行预测
    prediction = autoencoder.predict(new_ecg_data)

    # 计算重构误差
    mse_new = np.mean(np.power(new_ecg_data - prediction, 2), axis=1)

    # 判断是否异常
    if np.mean(mse_new) > threshold:
        return "Anomaly detected"
    else:
        return "Normal"

# 调用实时监测函数示例
new_ecg_data = load_new_ecg_data()  # 加载新的心电图数据
result = real_time_ecg_anomaly_detection(new_ecg_data)
print("Real-time anomaly detection result:", result)

# 代码示例：模型集成到医疗信息系统
from database import ECGRecord, save_anomaly_detection_result

def process_ecg_record(record_id):
    # 从数据库加载心电图记录
    ecg_data = ECGRecord.load(record_id)

    # 调用异常检测函数
    result = real_time_ecg_anomaly_detection(ecg_data)

    # 将检测结果保存到数据库
    save_anomaly_detection_result(record_id, result)