AI+医疗：使用神经网络进行医学影像识别分析 ⛵

    
|# 导入工具库||
|---|---|
|import numpy as np||
|import cv2||
|from glob import glob||
|from sklearn.model\_selection import train\_test\_split||
|import tensorflow as tf||
|from tensorflow.keras.callbacks import ModelCheckpoint, ReduceLROnPlateau||
|from tensorflow.keras.optimizers import Adam||
|from tensorflow.keras.metrics import Recall, Precision||

    
|# 读取X射线图像||
|---|---|
|x = cv2.imread(path, cv2.IMREAD\_COLOR)||
|x = cv2.resize(x, (width, height))||
|x = x/255.0||
|x = x.astype(np.float32)||
|return x||
|||
|# 读取标签蒙版||
|def maskread(path\_l, path\_r,width=512,height=512):||
|x\_l = cv2.imread(path\_l, cv2.IMREAD\_GRAYSCALE)||
|x\_r = cv2.imread(path\_r, cv2.IMREAD\_GRAYSCALE)||
|x = x\_l + x\_r||
|x = cv2.resize(x, (width, height))||
|x = x/np.max(x)||
|x = x > 0.5||
|x = x.astype(np.float32)||
|x = np.expand\_dims(x, axis=-1)||
|return x||

    
|"""加载与切分数据"""||
|---|---|
|images = sorted(glob(os.path.join(path, "CXR\_png", "*.png")))||
|masks\_l = sorted(glob(os.path.join(path, "ManualMask", "leftMask", "*.png")))||
|masks\_r = sorted(glob(os.path.join(path, "ManualMask", "rightMask", "*.png")))||
|split\_size = int(len(images) * split) # 9:1的比例切分||
|train\_x, val\_x = train\_test\_split(images, test\_size=split\_size, random\_state=42)||
|train\_y\_l, val\_y\_l = train\_test\_split(masks\_l, test\_size=split\_size, random\_state=42)||
|train\_y\_r, val\_y\_r = train\_test\_split(masks\_r, test\_size=split\_size, random\_state=42)||
|train\_x, test\_x = train\_test\_split(train\_x, test\_size=split\_size, random\_state=42)||
|train\_y\_l, test\_y\_l = train\_test\_split(train\_y\_l, test\_size=split\_size, random\_state=42)||
|train\_y\_r, test\_y\_r = train\_test\_split(train\_y\_r, test\_size=split\_size, random\_state=42)||
|||
|return (train\_x, train\_y\_l, train\_y\_r), (val\_x, val\_y\_l, val\_y\_r), (test\_x, test\_y\_l, test\_y\_r)||

    
|# tensor格式转换||
|---|---|
|def \_parse(x, y\_l, y\_r):||
|x = x.decode()||
|y\_l = y\_l.decode()||
|y\_r = y\_r.decode()||
|x = imageread(x)||
|y = maskread(y\_l, y\_r)||
|return x, y||
|x, y = tf.numpy\_function(\_parse, [x, y\_l, y\_r], [tf.float32, tf.float32])||
|x.set\_shape([512, 512, 3])||
|y.set\_shape([512, 512, 1])||
|return x, y||
|||
|# 构建tensorflow dataset||
|def tf\_dataset(X, Y\_l, Y\_r, batch=8):||
|dataset = tf.data.Dataset.from\_tensor\_slices((X, Y\_l, Y\_r))||
|dataset = dataset.shuffle(buffer\_size=200)||
|dataset = dataset.map(tf\_parse)||
|dataset = dataset.batch(batch)||
|dataset = dataset.prefetch(4)||
|return dataset||

    
|from tensorflow.keras.layers import Conv2D, BatchNormalization, Activation, MaxPool2D, Conv2DTranspose, Concatenate, Input||
|---|---|
|||
|# 一个卷积块结构||
|def conv\_block(input, num\_filters):||
|x = Conv2D(num\_filters, 3, padding="same")(input)||
|x = BatchNormalization()(x)||
|x = Activation("relu")(x)||
|||
|x = Conv2D(num\_filters, 3, padding="same")(x)||
|x = BatchNormalization()(x)||
|x = Activation("relu")(x)||
|||
|return x||
|||
|# 编码器模块||
|def encoder\_block(input, num\_filters):||
|x = conv\_block(input, num\_filters)||
|p = MaxPool2D((2, 2))(x)||
|return x, p||
|||
|# 解码器模块||
|def decoder\_block(input, skip\_features, num\_filters):||
|x = Conv2DTranspose(num\_filters, (2, 2), strides=2, padding="same")(input)||
|x = Concatenate()([x, skip\_features])||
|x = conv\_block(x, num\_filters)||
|return x||
|||
|# 完整的U-Net||
|def build\_unet(input\_shape):||
|inputs = Input(input\_shape)||
|||
|# 编码器部分||
|s1, p1 = encoder\_block(inputs, 64)||
|s2, p2 = encoder\_block(p1, 128)||
|s3, p3 = encoder\_block(p2, 256)||
|s4, p4 = encoder\_block(p3, 512)||
|||
|b1 = conv\_block(p4, 1024)||
|||
|# 解码器部分||
|d1 = decoder\_block(b1, s4, 512)||
|d2 = decoder\_block(d1, s3, 256)||
|d3 = decoder\_block(d2, s2, 128)||
|d4 = decoder\_block(d3, s1, 64)||
|||
|# 输出||
|outputs = Conv2D(1, 1, padding="same", activation="sigmoid")(d4)||
|||
|model = Model(inputs, outputs, name="U-Net")||
|return model||

    
|# IoU计算||
|---|---|
|def f(y\_true, y\_pred):||
|intersection = (y\_true * y\_pred).sum()||
|union = y\_true.sum() + y\_pred.sum() - intersection||
|x = (intersection + 1e-15) / (union + 1e-15)||
|x = x.astype(np.float32)||
|return x||
|return tf.numpy\_function(f, [y\_true, y\_pred], tf.float32)||
|||
|# Dice Loss定义||
|smooth = 1e-15||
|def dice\_coef(y\_true, y\_pred):||
|y\_true = tf.keras.layers.Flatten()(y\_true)||
|y\_pred = tf.keras.layers.Flatten()(y\_pred)||
|intersection = tf.reduce\_sum(y\_true * y\_pred)||
|return (2. * intersection + smooth) / (tf.reduce\_sum(y\_true) + tf.reduce\_sum(y\_pred) + smooth)||
|||
|def dice\_loss(y\_true, y\_pred):||
|return 1.0 - dice\_coef(y\_true, y\_pred)||

    
|# 超参数||
|---|---|
|lr = 1e-5||
|epochs = 30||
|model\_path = "models/model.h5"||
|||
|# 读取数据||
|dataset\_path = './NLM-MontgomeryCXRSet/MontgomerySet'||
|(train\_x, train\_y\_l, train\_y\_r), (val\_x, val\_y\_l, val\_y\_r), (test\_x, test\_y\_l, test\_y\_r) = load\_data(dataset\_path)||
|||
|# 训练集与验证集||
|train\_dataset = tf\_dataset(train\_x, train\_y\_l, train\_y\_r, batch=batch\_size)||
|val\_dataset = tf\_dataset(val\_x, val\_y\_l, val\_y\_r, batch=batch\_size)||
|||
|# 构建模型||
|model = build\_unet((512, 512, 3))||
|# 评估准则||
|metrics = [dice\_coef, iou, Recall(), Precision()]||
|# 编译模型||
|model.compile(loss=dice\_loss, optimizer=Adam(lr), metrics=metrics)||

    
|model . summary()||

    
|# 回调函数||
|---|---|
|ModelCheckpoint(model\_path, verbose=1, save\_best\_only=True),||
|ReduceLROnPlateau(monitor='val\_loss', factor=0.1, patience=5, min\_lr=1e-8, verbose=1)||
|]||
|||
|# 模型训练||
|history = model.fit(||
|train\_dataset,||
|epochs=epochs,||
|validation\_data=val\_dataset,||
|callbacks=callbacks||
|)||

    
|# 重新载入模型||
|---|---|
|with CustomObjectScope({'iou': iou, 'dice\_coef': dice\_coef, 'dice\_loss': dice\_loss}):||
|model = tf.keras.models.load\_model("/content/model.h5")||
|||
|||
|# 测试集预估||
|from tqdm import tqdm||
|import matplotlib.pyplot as plt||
|ct=0||
|||
|# 遍历测试集||
|for x, y\_l, y\_r in tqdm(zip(test\_x, test\_y\_l, test\_y\_r), total=len(test\_x)):||
|""" Extracing the image name. """||
|image\_name = x.split("/")[-1]||
|||
|# 读取测试图片集||
|ori\_x = cv2.imread(x, cv2.IMREAD\_COLOR)||
|ori\_x = cv2.resize(ori\_x, (512, 512))||
|x = ori\_x/255.0||
|x = x.astype(np.float32)||
|x = np.expand\_dims(x, axis=0)||
|||
|# 读取标签信息||
|ori\_y\_l = cv2.imread(y\_l, cv2.IMREAD\_GRAYSCALE)||
|ori\_y\_r = cv2.imread(y\_r, cv2.IMREAD\_GRAYSCALE)||
|ori\_y = ori\_y\_l + ori\_y\_r||
|ori\_y = cv2.resize(ori\_y, (512, 512))||
|ori\_y = np.expand\_dims(ori\_y, axis=-1) # (512, 512, 1)||
|ori\_y = np.concatenate([ori\_y, ori\_y, ori\_y], axis=-1) # (512, 512, 3)||
|||
|# 预估||
|y\_pred = model.predict(x)[0] > 0.5||
|y\_pred = y\_pred.astype(np.int32)||
|#plt.imshow(y\_pred)||
|||
|# 存储预估结果mask||
|save\_image\_path = "./"+str(ct)+".png"||
|ct+=1||
|y\_pred = np.concatenate([y\_pred, y\_pred, y\_pred], axis=-1)||
|sep\_line = np.ones((512, 10, 3)) * 255||
|cat\_image = np.concatenate([ori\_x, sep\_line, ori\_y, sep\_line, y\_pred*255], axis=1)||
|cv2.imwrite(save\_image\_path, cat\_image)||