10장 : 여러 데이터 소스를 통합 데이터셋으로 합치기

10장에서는 ...

• 원본 데이터 파일을 읽어 처리하고
• 데이터를 표현하는 파이썬 클래스를 구현하고
• 데이터를 파이토치에서 사용 가능한 포맷으로 변환하는 과정과
• 훈련 데이터와 검증 데이터의 시각화를 진행해봅니다 !

이번장에서는 어떻게 날것의 데이터셋을 모델에 맞게 ,pytorch에 맞게 텐서 형태로 만드는지를 작업해보는 파트다.

우리의 목표는 원본 CT 스캔 데이터와 데이터에 달아놓은 annotation 목롤으로 훈련 샘플을 만드는 것이다.

원본 CT 데이터 파일을 확인해보자.

CT 데이터 형식을 살펴보면,

• 메타데이터 header 정보가 포함된 .mhd 파일 과
• 3차원 배열을 만들 원본 데이터 바이트를 포함하는 .raw 파일 로 두가지다.

각 파일 이름은 시리즈 UID 라고 불리는 CT 스캔 단일 식별자로 시작되어있다.

CT 클래스는 두 파일을 읽어서 3차원 배열을 만들고, 환자 좌표계를 배열에서 필요로 하는 인덱스, 행, 열, 좌표로 바꿔주는 변환 행렬도 만든다.

여기서 인덱스, 행 , 열 을 앞으로 ( I, R, C ) 로 부르기로 한다.
코드는 _irc 로 끝나는 변수에 해당 !

자 이제 ANNOTATION 데이터도 살펴보자.

annotation 데이터에는 각 결절의 좌표 목록, 악성 여부, 그리고 해당 CT 스캔의 시리즈 UID 가 포함되어있다.

여기서 결절 좌표가 좌표계 변환 정보를 거치면 결절의 중심에 해당하는 복셀의 인덱스, 행, 열 정보가 생긴다.

(I,R,C) 좌표를 사용하면 CT 데이터의 작은 3차원 부분 단면을 얻어 모델에 대한 input 값으로 사용할 수 있다.

파이토치가 Dataset 서브클래스를 통해 얻고자 하는 것은 튜플이다 !

튜플 - 파이토치 텐서로 변환하는 과정까지 요구하는 것.

본 형식의 입력 변환을 위해 직접 고안한 알고리즘을 본 파트에서 사용하기도 하는데, 이를 피처 엔지니어링( feature engineering ) 이라고 한다.

데이터 로딩을 시작해보자 ( LUNA )

데이터 파싱을 진행해서 csv 파일 안의 내용 부터 확인해보자.

CT 정보가 포함된 candidates.csv 파일 내에 LUNA annotation 에서는, 결절 후보의 위치와 후보가 실제 결절인지의 여부를 알려주는 플래그가 존재한다.

파일 안 내용을 파싱해서 탐색해보는 코드는 다음과 같다.

$wc -l candidates.csv #파일의 행 개수 카운트

$ head data/part2/luna/candidates.csv #파일의 앞 부분 일부를 출력한디

$ grep ',1$' candidates.csv | wc -l # 결절이라 1로 끝나는 행 개수를 카운트한다.

• 데이터셋을 확인해보면, 전체 행 개수는 551,000이며, 각각은 seriesuid 와 (x,y,z) 좌표, 그리고 결절 상태를 의미하는 class 열로 구성되어있다.
• annotation.csv 파일에는 결절로 플래그된 후보들에 대한 정보가 포함되어있다.

훈련셋과 검증셋으로 나눠보자.

크기순으로 정렬한 후 매 N 번째에 대해 검증셋을 넣어 분포를 반영한 검증셋을 구성해보면, 이 데이터셋은 불친절하게도 annotation.csv 파일에서 제공하는 위치 정보는 candidates.csv 파일의 좌표와 정확하게 일치하기 않는다.

다음과 같이 데이터셋이 불일치한 경우라면, 무시하고 버려야한다 !

그렇다면 annotation 데이터와 후보 데이터를 합쳐볼까?

이 둘을 합치는 getCandidateInfoList 함수 를 만들어보자. 이때, 모델 훈련작업과 지저분한 데이터를 철저하게 분리하는 것이 좋은데, 이를 네임드 튜플(named tuple) 파일을 상단에 두고 사용하면 좋다.

from collections import namedtuple

CandidateInfoTuple = namedtuple(
    'CandidateInfoTuple',
    'isNodule_bool, diameter_mm, series_uid, center_xyz',
)

후보 정보만 디스크 공간 차지를 줄이기 위해서 requireOnDist_bool 파라미터를 활용해 뽑았다면 annotation.csv의 직경 정보를 합친다.

 diameter_dict = {}
    with open('data/part2/luna/annotations.csv', "r") as f:
        for row in list(csv.reader(f))[1:]:
            series_uid = row[0]
            annotationCenter_xyz = tuple([float(x) for x in row[1:4]])
            annotationDiameter_mm = float(row[4])

            diameter_dict.setdefault(series_uid, []).append(
                (annotationCenter_xyz, annotationDiameter_mm) 
                #여기서 annotation 정보는 series_uid 로 그룹화하여
                #두 파일에서 일치하는 행을 찾아내는 키로 사용한다. 
            )

두개를 합쳤다면, candidates.csv 정보를 사용해 전체 후보 리스트를 만든다.

   candidateInfo_list = []
    with open('data/part2/luna/candidates.csv', "r") as f:
        for row in list(csv.reader(f))[1:]:
        # for 문 돌며 같은 series_uid를 찾고 일치하는 경우를 찾았다면,
        #결절에 대한 직경 정보를 매칭했다고 간주함.
            series_uid = row[0]

            if series_uid not in presentOnDisk_set and requireOnDisk_bool:
                continue

            isNodule_bool = bool(int(row[4]))
            candidateCenter_xyz = tuple([float(x) for x in row[1:4]])

            candidateDiameter_mm = 0.0
            for annotation_tup in diameter_dict.get(series_uid, []):
                annotationCenter_xyz, annotationDiameter_mm = annotation_tup
                for i in range(3):
                    delta_mm = abs(candidateCenter_xyz[i] - annotationCenter_xyz[i])
                    if delta_mm > annotationDiameter_mm / 4:
                        break
                else:
                    candidateDiameter_mm = annotationDiameter_mm
                    break

            candidateInfo_list.append(CandidateInfoTuple(
                isNodule_bool,
                candidateDiameter_mm,
                series_uid,
                candidateCenter_xyz,
            ))
            
            #데이터 정렬 후 반환만 해주면 된다.
   		candidateInfo_list.sort(reverse=True)
    return candidateInfo_list

noduleInfo_list 의 튜플 멤버 순서는 다음과 같은 정렬로 만들어진다. 이렇게 데이터를 정렬하게 되면 일부 CT 단면들을 모아서 결절 직경에 대해 잘 분포된 실제 결절을 반영하는 덩어리를 얻을 수 있게된다.

개별 CT 스캔들을 로딩하자.

자, 다음 작업은 디스크에서 CT 데이터를 얻어 파이썬 객체로 변환하여 3차원 결절 밀도 데이터로 사용할 수 있도록 만드는 것이 목표이다.

데이터 파일 포맷이 무엇이든 넘파일 배열로 읽어들이기 위하여 SimpleTIK 를 사용한다.

class Ct:
    def __init__(self, series_uid):
        mhd_path = glob.glob(
            'data-unversioned/part2/luna/subset*/{}.mhd'.format(series_uid)
        )[0]

        ct_mhd = sitk.ReadImage(mhd_path)
        ct_a = np.array(sitk.GetArrayFromImage(ct_mhd), dtype=np.float32)

각 10개 서브셋에는 90개의 CT 스캔이 각각 들어있으며, CT 스캔은 .mhd 와 .raw 확장자를 가지는 두개의 파일로 나눠진다.

ct_a 는 3차원 배열로, 3개의 차원은 공간을 나타내고 하나는 밀도를 나타낸다.

자, 여기서 주의해야할 예외 데이터 처리

특정 CT 스캐너는 스캔 영역을 벗어난 복셀을 나타내기 위하여 밀도에 음의 값을 사용한다.

여기서 시야에 해당하는 영역을 위하여 값이 -1000HU 이하 일때는 버린다. 비슷하게 뼈나 금속에 해당하는 밀도값도 필요 없기 때문에 2g/cc(1000HU) 이상인 경우도 잘라낸다.

우리가 관심있는 경우의 종양은 1g/cc(0HU) 근처라는 것은 알아두자 !

 ct_a.clip(-1000, 1000, ct_a)

환자 좌표계를 사용하여 결절의 위치를 정하자.

우리는 중심이 잘 잡힌 후보 데이터로 정제를 진행해야하므로, 환자 좌표계가 날것으로 표현된 밀리미터 기반 좌표계인 (x,y,z) 로부터, CT 스캔 단면 데이터 배열에서 사용한 복셀 주소 기반 좌표계인 (I,R,C)로 좌표를 변환해야한다.

여기서 각각이 무엇을 뜻하는지 알아야하는데, 양의 X 값은 환자의 왼쪽, 양의 Y 값은 환자의 뒤쪽, 양의 Z 값은 환자의 머리 방향을 나타낸다. 이를 줄여 LPS 라고도 한다.

추가적으로 우리가 변환해야할 복셀 의 특징 또한 알아야하는데,
복셀은 정육면체가 아니기 때문에, 데이터를 정방형의 픽셀로 그려내게되면 왜곡된 이미지를 보이게 된다.

따라서 실제 비율로 보기 위해서 비율 계수(scale factor) 를 적용해야한다 !

각 CT는 파일의 메타데이터 내에 복셀의 크기를 밀리미터 단위로 정의하고 있으며, 리스트에서 이를 참조하기 위하여 ct_mhd.GetSpacing() 을 호출한다.

자, 그렇다면 밀리미터를 복셀 주소로 변환해볼까?

환자의 밀리미터 좌표와 (I,R,C) 배열 좌표 변환을 돕기 위하여 유틸리티 코드를 정의해보자. 이미 정해진 함수가 없기 때문에 자체 제작 함수를 사용하자. 함수는

• 좌표를 XYZ 체계로 만들기 위하여 IRC 에서 CRI 로 뒤집는다.

• 인덱스를 복셀 크기로 확대축소한다.

• 파이썬의 @ 를 사용하여 방향을 나타내는 행렬과 행렬 곱을 수행한다.

• 기준으로부터 오프셋을 더한다.

  복셀 크기의 경우, 네임드 튜플로 가지고 있음로 배열로 바꾸는 것까지 진행한다.

class Ct:
    def __init__(self, series_uid):
        mhd_path = glob.glob(
            'data-unversioned/part2/luna/subset*/{}.mhd'.format(series_uid)
        )[0]

        ct_mhd = sitk.ReadImage(mhd_path)
        ct_a = np.array(sitk.GetArrayFromImage(ct_mhd), dtype=np.float32)

        ct_a.clip(-1000, 1000, ct_a)

        self.series_uid = series_uid
        self.hu_a = ct_a

        self.origin_xyz = XyzTuple(*ct_mhd.GetOrigin())
        self.vxSize_xyz = XyzTuple(*ct_mhd.GetSpacing())
        self.direction_a = np.array(ct_mhd.GetDirection()).reshape(3, 3)

이리하여 각 후보의 센터를 환자 좌표에서 배열 좌표로 변환하기 위하여 CT 객체에 필요한 데이터를 넣는 일은 마무리되었다.

자, 추가적으로 데이터를 이쁘게 만들어보자.

getRawNodule 함수 는 LUNA CSV 데이터에서 명시된 환자 좌표계 (x,y,z) 로 표시된 중심 정보와 복셀 단위의 너비 정보도 인자로 전달받아 정육면체의 CT 덩어리와 배열 좌표로 변환된 후보의 중심값을 반환한다.

def getRawCandidate(self, center_xyz, width_irc):
        center_irc = xyz2irc(
            center_xyz,
            self.origin_xyz,
            self.vxSize_xyz,
            self.direction_a,
        )

        slice_list = []
        for axis, center_val in enumerate(center_irc):
            start_ndx = int(round(center_val - width_irc[axis]/2))
            end_ndx = int(start_ndx + width_irc[axis])

            assert center_val >= 0 and center_val < self.hu_a.shape[axis], repr([self.series_uid, center_xyz, self.origin_xyz, self.vxSize_xyz, center_irc, axis])

            if start_ndx < 0:
                # log.warning("Crop outside of CT array: {} {}, center:{} shape:{} width:{}".format(
                #     self.series_uid, center_xyz, center_irc, self.hu_a.shape, width_irc))
                start_ndx = 0
                end_ndx = int(width_irc[axis])

            if end_ndx > self.hu_a.shape[axis]:
                # log.warning("Crop outside of CT array: {} {}, center:{} shape:{} width:{}".format(
                #     self.series_uid, center_xyz, center_irc, self.hu_a.shape, width_irc))
                end_ndx = self.hu_a.shape[axis]
                start_ndx = int(self.hu_a.shape[axis] - width_irc[axis])

            slice_list.append(slice(start_ndx, end_ndx))

        ct_chunk = self.hu_a[tuple(slice_list)]

        return ct_chunk, center_irc

데이터셋 간단하게 구현해보자 !

자 이제는 dataset 인스턴스를 직접 구현해볼 차례 !

각 Ct 인스턴스는 모델을 훈련시키거나 효과를 검증할 때 사용하는 수백 개의 다양한 샘플이다. LunaDataset 클래스는 이러한 샘플들을 정규화하고 각 CT 결절은 flatten 작업 을 통해 어느 Ct 객체에서 가져온 샘플인지 상관없이 인출 가능하도록 단일 collection 으로 합쳐진다.

여기서 우리가 만들어야할 서브클래스는 파이토치 API 가 요구하는 두 함수만 구현하면 된다.

• 초기화 후에 하나의 상수값을 반환해야하는 __len__ 구현
• 인덱스 인자로 받아 훈련에서 사용할 샘플 데이터 튜플을 반환하는 __getitem__ 메소드

서브클래스들을 확인해볼까?

def __len__(self):
        return len(self.candidateInfo_list)

    def __getitem__(self, ndx):
        candidateInfo_tup = self.candidateInfo_list[ndx]
        width_irc = (32, 48, 48)

        candidate_a, center_irc = getCtRawCandidate(
            candidateInfo_tup.series_uid,
            candidateInfo_tup.center_xyz,
            width_irc,
        )

        candidate_t = torch.from_numpy(candidate_a)
        candidate_t = candidate_t.to(torch.float32)
        candidate_t = candidate_t.unsqueeze(0)

        pos_t = torch.tensor([
                not candidateInfo_tup.isNodule_bool,
                candidateInfo_tup.isNodule_bool
            ],
            dtype=torch.long,
        )

        return (
            candidate_t,
            pos_t,
            candidateInfo_tup.series_uid,
            torch.tensor(center_irc),
        )

__len__ 안에는 후보 리스트 하나하나는 샘플, 여기서 우리가 지켜야할 것은 __len__ 이 N값을 반환한다면 __getitem__ 은 0에서 N-1까지 입력값에 대해 유효한 아이템을 넘겨줘야한다.

__getitem__ 은 ndx 인자를 받아 샘플 배열, 결절 유무, series_uid, 후보 위치(I,R,C) 가 있는 샘플 튜플을 반환한다. __getitem__ 메소드에서 할 일은 이후 코드에서 사용할 데이터를 적절한 타입과 배열 차원으로 준비를 진행해줘야한다.

본격적으로 LUNA dataset을 다뤄보자.

• 먼저 getCtRawCandidate 함수로 후보 배열을 캐싱한다 - 시간 단축
• LunaDataset.__init__ 으로 데이터셋을 만든다. 여기서 isValSet_bool 파라미터를 통해 훈련 데이터나 검증 데이터만 둘지 둘다 둘지를 결정한다.
• 훈련/검증 세트를 분리한다. dataset의 N 번째 데이터들만 따로 모아서 검증용 서브셋을 만든다.

다음과 같이 데이터셋 정리가 끝났다면, 본격적으로 모델을 손봐볼까?