Chapter 8: 당뇨병성 망막병증(Diabetic Retinopathy) 탐지
이 장에서는 딥러닝이 실제 의료 진단에 어떻게 기여할 수 있는지 보여주는 대표적인 사례로, 망막 영상을 분석해 당뇨병성 망막병증(DR) 유무를 판별하는 프로젝트를 진행합니다.
프로젝트 구조 살펴보기
효율적인 분석을 위해 프로젝트를 세 가지 핵심 모듈로 구성했습니다: - data.py: 망막 이미지의 로딩과 전처리를 담당합니다. - model.py: 진단에 최적화된 딥러닝 모델의 구조를 정의합니다. - run.py: 모델의 학습부터 평가까지 전체 과정을 제어하는 실행 파일입니다.
상세 구현 내용
데이터 로딩 및 전처리
data.py의 load_images_DR 함수는 이미지를 중심에 맞춰 깔끔하게 자르고, 모델이 처리하기 좋게 \(512 \times 512\) 해상도로 크기를 조정합니다.
from __future__ import division
from __future__ import unicode_literals
import os
import logging
import deepchem
import numpy as np
import pandas as pd
"""
당뇨병성 망막병증(Diabetic Retinopathy) 영상 로더입니다.
"""
logger = logging.getLogger(__name__)
def load_images_DR(split="random", seed=None):
"""당뇨병성 망막병증(DR) 영상을 불러오는 로더입니다."""
data_dir = deepchem.utils.get_data_dir()
images_path = os.path.join(data_dir, "DR", "train")
label_path = os.path.join(data_dir, "DR", "trainLabels.csv")
if not os.path.exists(images_path) or not os.path.exists(label_path):
logger.warn(
"데이터를 찾을 수 없습니다. \n\
모든 영상 파일(.png)은 다음 폴더에 저장되어야 합니다: $DEEPCHEM_DATA_DIR/DR/train/,\n\
해당 라벨 파일은 $DEEPCHEM_DATA_DIR/DR/trainLabels.csv 경로에 저장되어야 합니다.\n\
데이터 접근에 대해서는 https://www.kaggle.com/c/diabetic-retinopathy-detection를 참조하세요."
)
image_names = os.listdir(images_path)
raw_images = []
for im in image_names:
if (
im.endswith(".jpeg")
and not im.startswith("cut_")
and "cut_" + im not in image_names
):
raw_images.append(im)
if len(raw_images) > 0:
cut_raw_images(raw_images, images_path)
image_names = [
p
for p in os.listdir(images_path)
if p.startswith("cut_") and p.endswith(".png")
]
all_labels = dict(zip(*np.transpose(np.array(pd.read_csv(label_path)))))
print("전체 영상 개수: %d" % len(image_names))
labels = np.array(
[all_labels[os.path.splitext(n)[0][4:]] for n in image_names]
).reshape((-1, 1))
image_full_paths = [os.path.join(images_path, n) for n in image_names]
classes, cts = np.unique(list(all_labels.values()), return_counts=True)
weight_ratio = dict(zip(classes, np.max(cts) / cts.astype(float)))
weights = np.array([weight_ratio[label[0]] for label in labels]).reshape((-1, 1))
dat = deepchem.data.ImageDataset(image_full_paths, labels, weights)
if split is None:
return dat
splitters = {
"index": deepchem.splits.IndexSplitter(),
"random": deepchem.splits.RandomSplitter(),
}
if seed is not None:
np.random.seed(seed)
splitter = splitters[split]
train, valid, test = splitter.train_valid_test_split(dat)
all_dataset = (train, valid, test)
return all_dataset
def cut_raw_images(all_images, path):
"""영상 전처리를 수행합니다:
(1) 망막을 포함한 중앙 영역을 정사각형으로 자릅니다.
(2) 해상도를 512 * 512로 조정합니다.
"""
print("처리할 영상 개수: %d" % len(all_images))
try:
import cv2
except: # noqa: E722
logger.warn("영상 전처리를 위해 OpenCV가 필요합니다.")
return
for i, img_path in enumerate(all_images):
if i % 100 == 0:
print("%d번째 영상 처리 중..." % i)
if os.path.exists(
os.path.join(path, "cut_" + os.path.splitext(img_path)[0] + ".png")
):
continue
img = cv2.imread(os.path.join(path, img_path))
edges = cv2.Canny(img, 10, 30)
coords = list(zip(*np.where(edges > 0)))
n_p = len(coords)
coords.sort(key=lambda x: (x[0], x[1]))
center_0 = int((coords[int(0.01 * n_p)][0] + coords[int(0.99 * n_p)][0]) / 2)
coords.sort(key=lambda x: (x[1], x[0]))
center_1 = int((coords[int(0.01 * n_p)][1] + coords[int(0.99 * n_p)][1]) / 2)
edge_size = min(
[center_0, img.shape[0] - center_0, center_1, img.shape[1] - center_1]
)
img_cut = img[
(center_0 - edge_size) : (center_0 + edge_size),
(center_1 - edge_size) : (center_1 + edge_size),
]
img_cut = cv2.resize(img_cut, (512, 512))
cv2.imwrite(
os.path.join(path, "cut_" + os.path.splitext(img_path)[0] + ".png"), img_cut
)의료 진단 모델 설계
정밀한 영상 진단을 목표로 설계된 딥러닝 모델의 구조를 살펴봅니다.
#!/usr/bin/env python2
# -*- coding: utf-8 -*-
"""
2018년 9월 10일 월요일 생성
@작성자: zqwu
"""
import deepchem as dc
import numpy as np
import tensorflow as tf
import tensorflow.keras.layers as layers
from sklearn.metrics import confusion_matrix, accuracy_score
class DRModel(dc.models.KerasModel):
def __init__(
self,
n_tasks=1,
image_size=512,
n_downsample=6,
n_init_kernel=16,
n_fully_connected=[1024],
n_classes=5,
augment=False,
batch_size=100,
**kwargs,
):
"""
매개변수(Parameters)
----------
n_tasks: int
작업(Task)의 수
image_size: int
입력 영상의 해상도 (정사각형)
n_downsample: int
2의 거듭제곱 비율로 다운샘플링 수행
n_init_kernel: int
첫 번째 합성곱 층(Convolutional layer)의 커널 크기
n_fully_connected: list of int
합성곱 연산 후 이어지는 완전 연결 층(FC layer)의 구조
n_classes: int
예측할 클래스 수 (분류 모드에서만 사용)
augment: bool
데이터 증강(Data augmentation) 사용 여부
"""
self.n_tasks = n_tasks
self.image_size = image_size
self.n_downsample = n_downsample
self.n_init_kernel = n_init_kernel
self.n_fully_connected = n_fully_connected
self.n_classes = n_classes
self.augment = augment
# 입력 플레이스홀더(Inputs placeholder)
self.inputs = tf.keras.Input(
shape=(self.image_size, self.image_size, 3), dtype=tf.float32
)
# 데이터 전처리 및 증강
in_layer = DRAugment(
self.augment, batch_size, size=(self.image_size, self.image_size)
)(self.inputs)
# 첫 번째 합성곱 층
in_layer = layers.Conv2D(
int(self.n_init_kernel), kernel_size=7, padding="same"
)(in_layer)
in_layer = layers.BatchNormalization()(in_layer)
in_layer = layers.ReLU()(in_layer)
# 최대 풀링(Max pooling)을 통한 다운샘플링
res_in = layers.MaxPool2D(pool_size=(3, 3), strides=(2, 2))(in_layer)
for ct_module in range(self.n_downsample - 1):
# 각 모듈은 잔차 합성곱 블록(Residual convolutional block)과
# 이어지는 합성곱 다운샘플링 층으로 구성됩니다.
in_layer = layers.Conv2D(
int(self.n_init_kernel * 2 ** (ct_module - 1)),
kernel_size=1,
padding="same",
)(res_in)
in_layer = layers.BatchNormalization()(in_layer)
in_layer = layers.ReLU()(in_layer)
in_layer = layers.Conv2D(
int(self.n_init_kernel * 2 ** (ct_module - 1)),
kernel_size=3,
padding="same",
)(in_layer)
in_layer = layers.BatchNormalization()(in_layer)
in_layer = layers.ReLU()(in_layer)
in_layer = layers.Conv2D(
int(self.n_init_kernel * 2**ct_module), kernel_size=1, padding="same"
)(in_layer)
res_a = layers.BatchNormalization()(in_layer)
res_out = res_in + res_a
res_in = layers.Conv2D(
int(self.n_init_kernel * 2 ** (ct_module + 1)),
kernel_size=3,
strides=2,
activation=tf.nn.relu,
padding="same",
)(res_out)
res_in = layers.BatchNormalization()(res_in)
# 최종 결과에 대한 최대 풀링
in_layer = layers.Lambda(lambda x: tf.reduce_max(x, axis=(1, 2)))(res_in)
regularizer = tf.keras.regularizers.l2(0.1)
for layer_size in self.n_fully_connected:
# 완전 연결 층
in_layer = layers.Dense(
layer_size, activation=tf.nn.relu, kernel_regularizer=regularizer
)(in_layer)
# 은닉층을 위한 드롭아웃(Dropout)
# in_layer = layers.Dropout(0.25)(in_layer)
logit_pred = layers.Dense(self.n_tasks * self.n_classes)(in_layer)
logit_pred = layers.Reshape((self.n_tasks, self.n_classes))(logit_pred)
output = layers.Softmax()(logit_pred)
keras_model = tf.keras.Model(inputs=self.inputs, outputs=[output, logit_pred])
super(DRModel, self).__init__(
keras_model,
loss=dc.models.losses.SparseSoftmaxCrossEntropy(),
output_types=["prediction", "loss"],
batch_size=batch_size,
**kwargs,
)
def DRAccuracy(y, y_pred):
y = np.argmax(y, 1)
y_pred = np.argmax(y_pred, 1)
return accuracy_score(y, y_pred)
def DRSpecificity(y, y_pred):
y_pred = (np.argmax(y_pred, 1) > 0) * 1
y = (y > 0) * 1
TN = sum((1 - y_pred) * (1 - y))
N = sum(1 - y)
return float(TN) / N
def DRSensitivity(y, y_pred):
y = np.argmax(y, 1)
y_pred = (np.argmax(y_pred, 1) > 0) * 1
y = (y > 0) * 1
TP = sum(y_pred * y)
P = sum(y)
return float(TP) / P
def ConfusionMatrix(y, y_pred):
y = np.argmax(y, 1)
y_pred = np.argmax(y_pred, 1)
return confusion_matrix(y, y_pred)
def QuadWeightedKappa(y, y_pred):
y = np.argmax(y, 1)
y_pred = np.argmax(y_pred, 1)
cm = confusion_matrix(y, y_pred)
classes_y, counts_y = np.unique(y, return_counts=True)
classes_y_pred, counts_y_pred = np.unique(y_pred, return_counts=True)
E = np.zeros((classes_y.shape[0], classes_y.shape[0]))
for i, c1 in enumerate(classes_y):
for j, c2 in enumerate(classes_y_pred):
E[c1, c2] = counts_y[i] * counts_y_pred[j]
E = E / np.sum(E) * np.sum(cm)
w = np.zeros((classes_y.shape[0], classes_y.shape[0]))
for i in range(classes_y.shape[0]):
for j in range(classes_y.shape[0]):
w[i, j] = float((i - j) ** 2) / (classes_y.shape[0] - 1) ** 2
re = 1 - np.sum(w * cm) / np.sum(w * E)
return re
class DRAugment(layers.Layer):
def __init__(
self,
augment,
batch_size,
distort_color=True,
central_crop=True,
size=(512, 512),
**kwargs,
):
"""
매개변수(Parameters)
----------
augment: bool
데이터 증강 사용 여부
batch_size: int
배치 내 영상 개수
distort_color: bool
색상 왜곡 무작위 적용 여부
central_crop: bool
중앙 중심의 무작위 자르기 적용 여부
size: int
입력 영상의 해상도 (정사각형)
"""
self.augment = augment
self.batch_size = batch_size
self.distort_color = distort_color
self.central_crop = central_crop
self.size = size
super(DRAugment, self).__init__(**kwargs)
def call(self, inputs, training=True):
parent_tensor = inputs / 255.0
if not self.augment or not training:
return parent_tensor
else:
def preprocess(img):
img = tf.image.random_flip_left_right(img)
img = tf.image.random_flip_up_down(img)
img = tf.image.rot90(img, k=np.random.randint(0, 4))
if self.distort_color:
img = tf.image.random_brightness(img, max_delta=32.0 / 255.0)
img = tf.image.random_saturation(img, lower=0.5, upper=1.5)
img = tf.clip_by_value(img, 0.0, 1.0)
if self.central_crop:
# 절단된 가우시안 분포로부터 샘플 비율을 추출합니다.
img = tf.image.central_crop(
img, np.clip(np.random.normal(1.0, 0.06), 0.8, 1.0)
)
img = tf.image.resize(
tf.expand_dims(img, 0), tf.convert_to_tensor(self.size)
)[0]
return img
return tf.map_fn(preprocess, parent_tensor)학습 및 실행
이제 run.py를 실행해 모델 학습을 시작하고 결과를 확인해 보겠습니다.
import deepchem as dc
# import numpy as np
# import pandas as pd
import os
import logging
from model import DRModel, DRAccuracy, ConfusionMatrix, QuadWeightedKappa
from data import load_images_DR
"""
2018년 9월 10일 월요일 생성
@작성자: zqwu
"""
RETRAIN = True
train, valid, test = load_images_DR(split="random", seed=123)
# 모델 정의 및 구축
model = DRModel(
n_init_kernel=32,
batch_size=32,
learning_rate=1e-5,
augment=True,
model_dir="./test_model",
)
if not os.path.exists("./test_model"):
os.mkdir("test_model")
if not RETRAIN:
os.system("sh get_pretrained_model.sh")
model.restore(checkpoint="./test_model/model-84384")
metrics = [
dc.metrics.Metric(DRAccuracy, mode="classification"),
dc.metrics.Metric(QuadWeightedKappa, mode="classification"),
]
cm = [dc.metrics.Metric(ConfusionMatrix, mode="classification")]
logger = logging.getLogger("deepchem.models.tensorgraph.tensor_graph")
logger.setLevel(logging.DEBUG)
if RETRAIN:
print("10 에포크 동안 모델 학습을 시작합니다.")
model.fit(train, nb_epoch=10, checkpoint_interval=1000)
print("훈련 데이터 지표 평가를 시작합니다.")
print(model.evaluate(train, metrics, n_classes=5))
print("검증 데이터 혼동 행렬(Confusion Matrix) 평가를 시작합니다.")
print(model.evaluate(valid, cm, n_classes=5))
print("테스트 데이터 혼동 행렬 평가를 시작합니다.")
print(model.evaluate(test, cm, n_classes=5))