用PyTorch构建CT扫描新冠肺炎分类器

您可以按照本教程的代码操作，并在ML Showcase中的免费GPU上运行它。

新冠肺炎继续对世界各地的医疗体系和经济造成严重破坏。50多万人死亡，1,140万人患病，10多亿人失业，新冠肺炎疫情可以说是21世纪最大的危机。我们还看到世界各国联合起来，以前所未有的规模抗击这一大流行-无论是加快疫苗试验、大规模生产口罩和呼吸机，还是采取巨大的经济刺激措施，以保持各国在封锁期间继续运转。

话虽如此，我相信机器学习社区可以发挥作用。事实上，这就是这个系列的全部内容。在最后的部分中，我概述了深度学习是如何被用来为新冠肺炎开发更好的测试方法的。我涵盖的所有文献都使用了从医院获得的医疗数据，这些数据在公共领域是不可获得的，这使得做任何形式的教程都很困难。然而，自那以后，这种情况发生了变化。

最近，加州大学圣地亚哥分校开源了一个数据集，其中包含新冠肺炎患者的肺部CT扫描图像，这在公共领域是第一个此类数据集。在这篇文章中，我们将使用PyTorch构建一个分类器，对患者进行肺部CT扫描，并将其分类为新冠肺炎阳性或阴性。

我们首先导入代码所需的模块，设置GPU，并设置TensorBoard目录以记录我们的培训指标。

将torchimport torch.nn作为nn从torch.utils.data导入数据集导入，DataLoader从torchvision导入作为转换从skimage.util导入montageimport osimport CV2导入随机导入matplotlib.pylot作为PLT导入torch.optim作为优化从PIL导入Imagefrom sklearn.metrics导入分类_报告、ROC_AUC_SCORE、ROC_CURVE、CONFUSION_matrixfrom torch.utils.optim作为优化从PIL导入Imagefrom sklearn.metrics导入分类_报告、ROC_AUC_SCORE、ROC_CURVE、CONFUSION_matrixfrom torch.util.。如果torch.cuda.is_available()否则"；CPU"；

我们将在GitHub上使用加州大学圣地亚哥分校提供的新冠肺炎CT扫描。此数据集包含从各种放射学/医学期刊(如MedRxiv、BioRxiv、NEJM、JAMA、Lancet)获取的图像。

我们首先克隆GitHub存储库以获取数据。从命令行运行：

下载数据后，用cd进入COVID-CT文件夹并解压包含图像的zip文件。

在我们开始构建分类器之前，让我注意一下数据的结构。我们有新冠肺炎阳性患者的扫描阳性类，而阴性类包括健康患者和其他(非新冠肺炎)疾病可能导致肺部混浊的患者。

为了训练一个稳健的分类器，我们还必须了解非新冠肺炎患者的信息。这一点很重要，因为医生从来不会直接派人去做CT扫描。事实上，肺炎是一种临床诊断，接受CT扫描的人很可能已经患上了其中一种呼吸系统疾病，如病毒性/细菌性肺炎/链球菌性肺炎等。我们很少看到健康的病人被送去做CT扫描。

因此，一个实用的分类器必须区分，比方说新冠肺炎引起的肺炎和其他类型的肺炎。然而，这个数据集中的负值类别是混淆的，它包含健康的肺部，以及患有其他疾病(如癌症)的患者的肺部。那么这么说有什么意义呢？关键是你应该把这个量词当作一个用于教育目的的量词。然而，任何您想要放在野外的分类器都需要更多的差异化数据。

COVID_FILES_PATH='；Images-processed/CT_COVID/'；covid_files=[os.path.join(COVID_FILES_PATH，x)for x in os.listdir(Covid_Files_Path)]covid_images=[cv2.imread(X)for x in Random.sample(covid_files，5)]plt.figure(figsize=(20，10))Columns=5for i，image in Enumerate(Covid_Image)：plt.sublot(len(covid_images，5))plt.figure(figsize=(20，10))column=5for i，image in枚举(Covid_Image)：plt.sublot(len(covid_image。

类似地，我们可以通过将COVID_FILES_PATH变量的值更改为“Images-Proceded/CT_NonCOVID”来查看非日冕病例的随机样本。

数据集分为三个部分：训练集(425个示例)、验证集(118个示例)和测试集(203个示例)。此拆分的信息已在文件夹Data-Split文件夹中提供。*此文件夹包含解释每个拆分所属文件的文本文件。

我们编写一个函数来读取这些文件，并将它们放入字符串列表中。

def read_txt(Txt_Path)：将open(Txt_Path)设置为f：ines=f.readines()txt_data=[line.strie()表示行中的行]返回txt_data。

class CovidCTDataset(DataSet)：def__init__(self，root_dir，class，covid_files，non_covid_files，Transform=None)：self.root_dir=root_dir self.class=class self.files_path=[non_covid_files，covid_files]self.image_list=[]#从数据拆分文本文件中读取文件covid_files=read_txt(Covid_Files)non_covid_files=read_txt(Non_Covid_Files)#将正负文件合并成CLS_INDEX的累加文件列表(len(self.class))：class_files=[[os.path.join(self.root_dir，self.class[。cls_index]\for x in read_txt(self.files_path[cls_index])]self.image_list+=class_files self.Transform=变换def__len__(Self)：返回镜头(self.image_list)def__getitem_(self，idx)：path=self.image_list[idx][0]#读取image=Image.open(Path).Convert('；rgb'；)#如果是self.change：image=self.change(Image)label=int(self.image_list[IDX][1])data={'；img'；：image，'；label'；：label，'；path'；：path}返回数据。

数据集返回包含图像张量、标签张量和批次中包含的图像路径列表的字典。

随机裁剪大小从图像尺寸的50%到100%，以及宽高比的随机范围，从原始长宽比的75%到133%。最后，裁剪的大小调整为224×224。

Normalize=转换。Normalize(Mean=[0，0，0]，Std=[1，1，1])Train_Transformer=Transforms。Compose([Transforms.Resize(256)，Transforms.RandomResizedCrop((224)，Scale=(0.5，1.0))，Transforms.RandomHorizontalFlip()，Transforms.ToT()，Normalize])Val_Transformer=Transforms。Compose([Transforms.Resize((224,224。

定义了DataSet和DataLoader类之后，现在让我们实例化它们。对于非COVID病例，我们使用标签0，而对于COVID阳性病例，我们使用1。

BatchSize=8Training Set=CovidCTDataset(root_dir='；Images-processed/'；，CLASS=['；CT_NONCOVID'；，'；CT_COVID'；]，covid_files='；Data-split/COVID/trainCT_COVID.txt'；，non_covid_files='；Data-split/NonCOVID/trainCT_NonCOVID.txt'；，转换=TRAIN_TRANSFER)阀集=CovidCTDataset(root_dir='；Images-processed/'；，CLASS=['；CT_NON COVID'；，'；CT_COVID'；]，covid_files='；Data-split/COVID/valCT_COVID.txt'；，NON_COVID_FILES='；Data-split/NonCOVID/valCT_NonCOVID.txt'；，变换=VAL_转换器)测试集=CovidCTDataset(root_dir='；Images-processed/'；，CLASS=['；CT_NONCOVID'；，'；CT_COVID'；]，COVID_FILES='；DATA-SPLIT/COVID/TESTCT_COVID.txt'；，non_covid_files='；Data-split/NonCOVID/testCT_NonCOVID.txt'；，Transform=VAL_Transform)TRAIN_LOADER=数据加载器(培训集，BATCH_SIZE=批大小，DROP_LAST=FALSE，SHUFFLE=真)VAL_LOADER=数据加载器(VALSET，BATCH_SIZE=批大小，DROP_LAST=FALSE，SHUFFLE=FALSE)TEST_LOADER=数据加载器(测试集，BATCH_SIZE=批大小，DROP_。

正如我们在第1部分中介绍的，准确性可能不足以确定分类器的有效性。因此，我们需要计算灵敏度、特异度、ROC下的面积等指标。我们编写函数COMPUTE_METRICS来计算这些指标和其他一些量，这些量将在以后的分析中有用。

def COMPUTE_METRICS(MODEL，TEST_LOADER，PLOT_ROC_CURE=FALSE)：Model.eval()val_Loss=0VAL_CORRECT=0 Criteria=nn.CrossEntropyLoss()SCORE_LIST=torch.TEXTER([]).TO(DEVICE)pred_LIST=torch.TEXTER([]).TO(DEVICE).Long()target_LIST=torch.TEXTER([]).TO()PATH_。目标=数据['；img'；].to(设备)，data['；Label'；].to(设备)路径=data['；路径'；]path_list.add(Path)#用torch.no_grad()计算损失。no_grad()：OUTPUT=MODEL(IMAGE)#LOG LOSS VAL_LOSS+=Criteria(output，target.long()).Item()#计算正确分类的示例个数pred=output.argmax(dim=1，Keepdim=True)val_校正+=pred.eq(target.long().view_as(pred)).sum().item()#簿记分数_列表=torch.cat([SCORE_LIST，nn.Softmax(dim=1))。1].squze()])pred_list=torch.cat([pred_list，pred.squeeze()])target_list=torch.cat([target_list，target.squeeze()])分类度量=分类报告(target_list.tolist()，pred_list.tolist()，target_ames=['；CT_NONCOVID'；，'；CT_COVID'；]，OUTPUT_DICIT=TRUE)#敏感度是对正类别敏感度的召回=classification_metrics['；CT_COVID'；]['；recall'；]#特定性是对负类特定性的召回=classification_metrics['；CT_NonCOVID'；]['；recall'；]#准确性=分类度量['；准确性'；]#混淆矩阵conf_Matrix=Conflation_Matrix(target_list.tolist()，pred_list.tolist())#ROC SCORE ROC_SCORE=ROC_AUC_SCORE(target_list.tolist()，core_list.tolist())#如果Plot_Roc_curve：fpr，tpr，_=roc_curve(target_list.tolist()，core_list.tolist()plt.)，则绘制ROC曲线。.format(ROC_SCORE))plt.Legend(loc='；Best'；)plt.xlabel('；假阳性率'；)plt.ylabel('；真阳性率'；)plt.show()#将值metrics_dict={"；准确性"；：准确性，"；敏感性"；：敏感性，"；特异性&34；：特异性，"；ROC_。混淆矩阵"；：conf_Matrix，"；验证丢失"；：val_oss/len(Test_Loader)，"；core_list"；：core_list.tolist()，"；pred_list"；：prd_list.tolist()，"；target_list"；：target_list.tolist()，"；path"；：path_list}返回元。

我们现在定义我们的模型。我们使用具有批归一化的预先训练的VGG-19作为我们的模型。然后，我们用输出端有2个神经元的线性层替换它的最后一个线性层，并在我们的数据集上执行转移学习。

现在，你也可以尝试其他型号，如ResNet，DenseNet等，特别是如果你正在寻找更轻的型号，因为VGG-19比ResNet或DenseNet有更多的参数。我选择使用VGG是因为它通常会带来更直观的激活图。

如果要使用其他模型，请确保将最后一层替换为具有两个输出。

现在我们设置训练超参数。我们使用的初始学习率为0.01。我们使用动量值为0.9的随机梯度下降。

我们实现了一个名为EarlyStopping的类，它保持损失和准确率的运行平均值。*这将有助于我们实施，嗯，你猜对了-提前停止。

这一类保持损失和准确度的移动平均值。如果度量没有改进超过由耐心定义的设置的纪元数，则方法Stop返回：

请注意，对于某个指标，术语“耐心”的用法已经用完，这意味着该指标在设定的纪元数内没有得到改善。

从集合中导入dequecclass EarlyStopping(Object)：def__init__(self，patience=8)：Super(EarlyStopping，self).__init__()self.patience=Patience self.previous_losest=int(1e8)self.previous_accuracy=0 self.init=false self.accuracy_reduce_iters=0 self.oss_increase_iters=0 self.Best_run_accuracy=0 self.Best_run_accuracy=。Accuracy)：#如果不是self.init，则计算移动平均值。init：Running_Loss=Loss Running_Accuracy=Accuracy self.init=True Else：Running_Loss=0.2*Loss+0.8*self.Previous_Accuracy=0.2*Accuracy+0.8*self.Previous_Accuracy#如果Running_Accuracy<，检查运行精度是否已提高到目前记录的最佳运行精度之外；self.Best_Running_Accuracy：self.Accuracy_Reduce_iters+=1否则：self.Best_Running_Accuracy=Running_Accuracy self.Accuracy_Reduce_iters=0#如果Running_Loss>，检查运行损耗是否比迄今记录的最佳运行损耗有所下降；self.Best_Running_Loss：self.oss_increase_iters+=1否则：self.Best_Running_Loss=Running_Loss self.oss_increase_iters=0#记录当前精度和损耗本身。Previous_Accuracy=Running_Accuracy self.Previous_Loss=Running_Loss def Stop(SELF)：#COMPUTE THRESHOLDS_THRESHOLD=self.Accuracy_Reduce_ITERS>；self.Patience Loss_Threshold=self.Loss_Increase_iters&Gt；self.peratience#如果Accuracy_Threshold和Loss_Threshold：如果Accuracy_Threshold和Loss_Threshold：如果Accuracy_Threshold返回1：如果Loss_Threshold返回2如果Loss_Threshold：Return 3返回0 def Reset(Self)：#Reset self.accuracy_Reduce_iters=0 self.oss_increase_iters=0 arly_stopper=EarlyStopping(Patience=5)，则返回代码。

如果对运行验证损失的耐心耗尽，而不是对运行准确性的耐心耗尽，我们会将学习率乘以0.1。如果对运行验证损失和运行准确性的耐心都已耗尽，我们将停止培训。

这种策略的原因在于交叉熵损失的性质，其中较高的验证损失不一定对应于较低的精度。为什么？因为交叉熵损失的一个微妙之处在于它更喜欢高置信度预测。因此，对其预测不太有信心的更准确的模型可能比精度较低但非常有信心的预测的模型有更高的损失。因此，只有当精度也停止增加时，我们才会做出停止的决定。

如您所见，我使用的批处理大小为8。但是，要获得好的结果，您必须使用更高的批处理大小，比如64或128。我的RTX 2060只能容纳8个批次大小。要实质上实现64个批次更新，我们可以在8次迭代(8(批次大小)*8(迭代次数)=64)上累积梯度，然后仅在那时执行渐变更新。执行此操作的基本模板非常简单。

我们将损失除以8，因为我们要为8次迭代添加更新，并且我们需要重新调整损失。

以下是训练循环的代码。这是一大段代码，所以我添加了注释，这样你就可以很容易地跟上了。

BEST_MODEL=MODEBEST_VAL_SCORE=0criteria=nn.CrossEntropyLoss()对于范围(60)中的纪元：model.Train()Train_Loss=0对于ITER_Num，数据在枚举(Train_Loader)中：image，target=data['；img'；].to(Device)，data['；label'；].to(Device)#计算损失输出=model(Image)Loss=Criteria(output，target.long())/8#记录损失Train_Loss+=loss.Item()loss.backward()#如果ITER_Num%8==0：Optimizer.step()Optimizer.zero_grad()#计算正确分类的示例的数量pred=output.argmax(dim=1，)#执行梯度更新，如果ITER_num%8==0：Optimizer.step()Optimizer.ero_grad()#计算正确分类的示例的数量pred=output.argmax(dim=1，Keepdim=TRUE)TRAIN_CORRECT+=pred.eq(target.long().view_as(pred)).sum().item()#计算并打印性能指标METRICS_DICT=COMPUTE_METRICS(MODEL，VAL_LOADER)PRINT('；-纪元{}Iteration{}-'；.format(epoch，iter_Num)打印("；准确性\t{：.3f}"；.format(metrics_dict['；Accuracy'；]))打印("；敏感度\t{：.3f}"；.format(METRICS_DICT['；敏感度])打印("；Specific\t{：.3f}"；.format(metrics_dict['；Specificity'；]))Print("；ROC\t{：.3f}"；.format(metrics_dict['；Roc_score'；]))Print("；Valal Lost\t{}"；.Format(METRICS_DICT["；VALL LOSS\t{}"；.Format(METRICS_DICT["；Validation Lost"；]))print("；--"；)#如果METRICS_DICT['；精度'；]>；，则以最佳验证精度保存模型。BEST_VAL_SCORE：torch.save(model，"；Best_model.pkl"；)BEST_VAL_SCORE=METRICS_DICT['；Accuracy'；]#打印纪元图的训练数据指标('；\n培训绩效纪元{}：平均损失：{：.4f}，精度：{}/{}({：.0f}%)\n'；.format(EPOCH，TRAIN_LOSS/LEN(TRAIN_LOADER.DataSet)，TRAIN_REGRECT，LEN(TRAIN_LOADER.DataSet)，100.0*TRAIN_CORPORT/LEN(tra

加载具有最佳验证精度的模型，该模型存储为Best_Model.pkl。要加载它，请使用model=torch.load('；Best_mod.pkl)。

这里提供了预先培训的模型。下载模型并使用模型=torch.load('；pretrained_covid_model.pkl'；).。从此处下载预先培训的模型。

加载模型后，您可以使用以下代码计算性能指标。

MODEL=torch.load("；pretrained_covid_model.pkl"；)METRICS_DICT=COMPUTE_METRICS(MODEL，TEST_LOADER，PLOT_ROC_CURE=TRUE)打印('；-测试性能-'；)print("；精度\t{：.3f}"；.format(metrics_dict['；Accuracy'；]))print("；Sensitivity\t{：.3f}"；.format(metrics_dict['；Sensitivity'；]))print("；Specificity\t ROC{：.3f}"下的{：.3f}"；.format(metrics_dict['；Specificity'；]))print("；Area；.format(metrics_dict['；Roc_score'；]))print("；--"；)。

conf_Matrix=metrics_dict["；混淆矩阵"；]ax=plt.sublot()sns.heatmap(conf_Matrix，annot=True，ax=ax，cmap='；Blues'；)；#annot=True注释单元格#labels，title和ticksax.set_xlabel('；预计标签'；)；ax.set_ylabel('；True labels&#。CoViD&39；，'；NonCoViD&39；])；ax.yaxis.set_ticklabels(['；CoViD&39；，'；NonCoViD'；])；

现在我们来看看我们的模型所犯的错误。我们首先得到错误分类样本的索引。然后，我们查看分配给错误分类示例的分数，并绘制直方图。

目标=np.array(metrics_dict['；target_list'；])preds=np.array(metrics_dict['；pred_list'；])scores=np.array(metrics_dict['；score_list'；])misclassified_indexes=np.non零(目标！=前)错误分类_。

..