{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":71549,"databundleVersionId":8561470,"sourceType":"competition"},{"sourceId":6158,"sourceType":"modelInstanceVersion","modelInstanceId":4608,"modelId":2797},{"sourceId":64765,"sourceType":"modelInstanceVersion","modelInstanceId":54020,"modelId":74163},{"sourceId":64795,"sourceType":"modelInstanceVersion","modelInstanceId":54048,"modelId":74163}],"dockerImageVersionId":30733,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# RSNA 2024 Lumbar Spine Degenerative Classification","metadata":{}},{"cell_type":"markdown","source":"## Starter Notebook for Pytorch and Deep learning techniques","metadata":{}},{"cell_type":"markdown","source":"What does this notebook contains?\n\n* Data organized in an understandable and easy to use way\n* A pretrained EfficientNetV2 for inference","metadata":{}},{"cell_type":"code","source":"import numpy as np\nimport pandas as pd\n\nimport matplotlib.pyplot as plt\nimport matplotlib.patches as patches\nfrom matplotlib import animation, rc\nimport seaborn as sns\n\nimport pydicom as dicom # dicom\nfrom pydicom.pixel_data_handlers.util import apply_voi_lut\n\nimport os\nimport time\nimport glob\nimport json\nimport collections\nimport torch\nimport torch.nn as nn","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# read data\ntrain_path = '/kaggle/input/rsna-2024-lumbar-spine-degenerative-classification/'\n\n# 파일 구조 : study - series - instance\ntrain  = pd.read_csv(train_path + 'train.csv')\nlabel = pd.read_csv(train_path + 'train_label_coordinates.csv')\ntrain_desc  = pd.read_csv(train_path + 'train_series_descriptions.csv')\ntest_desc   = pd.read_csv(train_path + 'test_series_descriptions.csv')\nsub         = pd.read_csv(train_path + 'sample_submission.csv')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"train shape :\", train.shape)\nprint(\"train label shape :\", label.shape)\nprint(\"train descriptions shape :\", train_desc.shape)\nprint(\"test descriptions shape :\", test_desc.shape)\nprint(\"submission shape :\", sub.shape)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# study_id, series_id를 바탕으로 instance 이미지 파일에 접근할 수 있도록 경로를 생성해주는 함수\n# 이미지 파일이 있는 폴더 경로를 입력으로 받고, 해당 경로에서 df의 study_id, series_id를 하위경로명으로 가지는 폴더로 접근.\n# 각 시리즈별 폴더에 속한 파일명을 조회한 뒤, 이를 붙여 전체 파일경로를 만들어 image_paths에 저장한 뒤 반환.\n\ndef generate_image_paths(df, data_dir):\n    image_paths = [] #이미지 파일 경로를 저장할 리스트\n    for study_id, series_id in zip(df['study_id'], df['series_id']): #인수로 받은 데이터 프레임의 study_id, series_id열을 tuple로 묶어서 조회\n        study_dir = os.path.join(data_dir, str(study_id)) #인수로 받은 경로에 study_id를 폴더명으로 가지는 하위경로를 study_dir에 저장\n        series_dir = os.path.join(study_dir, str(series_id)) #study_dir에서 series_id를 폴더명으로 가지는 하위경로를 series_dir에 저장\n        images = os.listdir(series_dir) #series_dir의 파일목록 조회\n        image_paths.extend([os.path.join(series_dir, img) for img in images]) #파일목록을 series_dir에 추가하여 image_paths에 저장\n    return image_paths\n\n#train_desc의 study_id, series_id열을 조회하여, /train_images 폴더에서 해당 스터디-시리즈에 포함된 이미지 경로를 추출\ntrain_image_paths = generate_image_paths(train_desc, f'{train_path}/train_images') \n#test_desc의 study_id, series_id열을 조회하여, /test_images 폴더에서 해당 스터디-시리즈에 포함된 이미지 경로를 추출\ntest_image_paths = generate_image_paths(test_desc, f'{train_path}/test_images')\n\n# 경로가 잘 생성되었음을 확인\ntrain_image_paths[:5]","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"n of study \\t:\", len(train))\nprint(\"n of series \\t:\", len(train_desc))\nprint(\"n of images \\t:\", len(train_image_paths))\nprint(\"n of labels \\t:\", len(label)) \n\nprint('\\n모든 image(instance)에 label이 부여된 것은 아님')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# DICOM 이미지 한장을 열어보기\n# DICOM : 의료 이미징 정보를 저장하고 전송하기 위한 표준 형식\n# pydicom 라이브러리로 읽어들이고, matplotlib을 통해 화면에 나타낼 수 있음\nplt.figure(figsize=(15,5)) \nds = dicom.dcmread(train_image_paths[30]) #30번째 이미지\nplt.imshow(ds.pixel_array, cmap = plt.cm.bone)\nprint(ds.pixel_array.shape) #(320,320) 해상도의 이미지","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#DICOM 파일에 포함된 메타데이터\nds.dir","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# DICOME 이미지를 열고 표시하기 위한 함수\ndef display_dicom_images(image_paths, n):\n    plt.figure(figsize=(12,4)) #이미지 크기\n    for i, path in enumerate(image_paths[:n]): #n번째 이미지까지 조회 - enumerate()은 iterable 객체를 받아 인덱스와 요소를 함께 반환\n        ds = dicom.dcmread(path) \n        plt.subplot(1, 3, i+1)\n        plt.imshow(ds.pixel_array, cmap = plt.cm.bone)\n        plt.title(f\"Image{i+1}\") #\n        plt.axis('off')\n    plt.show()\n\n# train 이미지의 첫번째 사진 3개 출력\ndisplay_dicom_images(train_image_paths,3)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_desc[train_desc['study_id'] == 10728036] # 같은 방향 사진이 2장 있는 경우도 있음","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"label","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#이미지를 label에 저장된 병변의 좌표와 함께 표시\n\n#이미지 경로와 레이블을 받아 출력하는 함수\ndef display_dicom_with_coordinates(image_paths, label_df, exact_image):\n    \n    # len(image_paths) 개의 서브플롯을 가로로 나란히 생성하고, 각각의 서브플롯에 DICOM 이미지를 표시\n    fig, axs = plt.subplots(1, len(image_paths), figsize=(12,4)) #fig는 전체그림, axs는 개별 서브플롯\n\n    for index, path in enumerate(image_paths):\n\n        # 이미지 경로로부터 다시 study_id, series_id 추출\n        study_id = int(path.split('/')[-3]) #study_id 추출\n        series_id = int(path.split('/')[-2]) #series_id 추출\n\n        # DICOM 이미지 오픈\n        ds = dicom.dcmread(path)\n\n        # DICOM 이미지 표시\n        axs[index].imshow(ds.pixel_array, cmap=plt.cm.bone)\n        axs[index].set_title(f'Study ID: {study_id}, Series ID: {series_id}')\n        axs[index].axis('off')\n\n        # label 데이터프레임에서 이미지와 동일한 study_id, series_id를 가지는 행을 추출 \n        # 같은 시리즈의 병변 좌표를 모두 출력(exact_image == False)\n        if exact_image == False:\n            filtered_labels = label_df[(label_df['study_id'] == study_id) & \n                                       (label_df['series_id'] == series_id)]\n        # 이미지와 동일한 instance_number 조건까지 추가하면 정확히 해당 이미지의 label만 출력(exact_image == True)\n        elif exact_image == True:\n            instance_number = ds.InstanceNumber # DICOM 파일에서 instance_number 추출\n            filtered_labels = label_df[(label_df['study_id'] == study_id) & \n                                       (label_df['series_id'] == series_id) &\n                                       (label_df['instance_number'] == instance_number)]\n        else:\n            raise ValueError('Invalid exact_image value')\n\n        # 지정된 조건에 따라 filtered_labels에 저장된 행들을 조회하며 병변의 좌표 출력\n        # exact_image == False으로 여러 행이 존재할 경우, 여러 마커를 출력\n        for _, row in filtered_labels.iterrows(): \n            axs[index].plot(row['x'], row['y'], 'ro', markersize=3)\n        \n    plt.tight_layout()\n    plt.show()\n\n# display_dicom_with_coordinates(train_image_paths[0:3], label) - 정상작동함을 확인\n\n# 폴더 경로로부터 인스턴스 파일들을 불러오는 함수\ndef load_dicom_files(series_folder_path):\n    # series_path_to_folder가 유효한 디렉토리인지 확인\n    if os.path.isdir(series_folder_path):\n        # series_path_to_folder에 속한 파일들을 불러오되, 확장자가 '.dcm'으로 되어있는지 확인\n        files = [os.path.join(series_folder_path, f) \n                 for f in os.listdir(series_folder_path) \n                 if f.endswith('.dcm')]    \n        # basename()으로 전체 파일경로에서 파일명을 추출하고, splitext()으로 확장자를 제거한 뒤, '-'를 기준으로 나누어 인스턴스 번호를 추출하고, 이를 기준으로 정렬\n        files.sort(key=lambda x: int(os.path.splitext(os.path.basename(x))[0].split('-')[-1]))\n        return files\n    return []\n\n# DICOM 이미지를 병변 좌표와 함께 표시\nstudy_id = train['study_id'][0] # 예시로 첫번째 study_id를 사용해보았음\nstudy_folder = f'{train_path}/train_images/{study_id}' # study_folder 경로를 만들고\n\nimage_paths = [] # image_paths 초기화\nfor series_folder in os.listdir(study_folder): # study_folder 내 series_folder들에 대해\n    series_folder_path = os.path.join(study_folder, series_folder) # series_folder 경로를 따오고\n    if not os.path.isdir(series_folder_path) : continue # 디렉토리가 아닌 경우, 다음 항목으로 넘어감\n    dicom_files = load_dicom_files(series_folder_path) # series_folder 경로로부터 인스턴스 파일들을 경로들을 불러온 뒤\n    if dicom_files:\n        image_paths.append(dicom_files[0]) # 시리즈의 첫 인스턴스 파일 경로를 image_paths에 저장\n\ndisplay_dicom_with_coordinates(image_paths, label, False) # image_paths에 저장된 경로의 이미지를 label과 함께 출력","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Data Preprocessing","metadata":{}},{"cell_type":"code","source":"# train 데이터 프레임의 모양을 다시 확인 -> wide format\n# 증상명(spinal_canal_stenosis)과 부위(l1_l2)를 분리해 'tidy'하게 바꿔줄 필요가 있음\n# '_'이 여러번 등장하고, 규칙이 복잡하여 melt 메서드를 사용하기 보다 그냥 함수를 정의하는게 나음\ntrain.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# long format의 행을 받아 tidy한 데이터 프레임을 반환하는 함수\ndef reshape_row(row):\n    data = {'study_id':[], 'condition':[], 'level':[], 'severity':[]} # tidy한 데이터 프레임의 형태\n\n    for column, value in row.items(): # 행의 각 열을 조회하여 열 이름과 값을 튜플 형태로 불러옴 \n        if column not in ['study_id','series_id', 'instance_number', 'x', 'y', 'series_description']: # tidy elements : study_id는 그대로\n            parts = column.split('_') # spinal_canal_stenosis_l1_l2 -> ['spinal', 'canal', 'stenosis', 'l1', 'l2']\n            condition = ' '.join([word.capitalize() for word in parts[:-2]]) # ['spinal', 'canal', 'stenosis'] -> 'Spinal Canal Stenosis'\n            level = parts[-2].capitalize() + '/' + parts[-1].capitalize() # ['l1', 'l2'] -> 'L1/L2'\n            data['study_id'].append(row['study_id'])\n            data['condition'].append(condition)\n            data['level'].append(level)\n            data['severity'].append(value)\n\n    return pd.DataFrame(data)\n\n# train 데이터프레임의 모든 행에 대해 reshape_row()를 적용한 결과를 결합하여 new_train_df에 저장\nnew_train_df = pd.concat([reshape_row(row) for _, row in train.iterrows()], ignore_index=True)\nnew_train_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"new_train_df","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 각 데이터 프레임의 컬럼을 다시 확인\n# new_train_df, label, train_desc를 병합해야 함\nprint(\"\\nColumns in new_train_df:\")\nprint(\", \".join(new_train_df.columns))\n\nprint(\"\\nColumns in label:\")\nprint(\", \".join(label.columns))\n\nprint(\"\\nColumns in train_desc:\")\nprint(\", \".join(train_desc.columns))\n\nprint(\"\\nColumns in sub:\")\nprint(\", \".join(sub.columns))","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# new_train_df와 label의 병합\nmerged_df = pd.merge(new_train_df, label, on=['study_id', 'condition', 'level'], how='inner')\n\n# merged_df(new_train_df+label)와 train_desc의 병합\nfinal_merged_df = pd.merge(merged_df, train_desc, on=['series_id','study_id'], how='inner')\n\n# 병합결과 확인\nfinal_merged_df.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"pd.crosstab(final_merged_df['condition'], final_merged_df['series_description'])","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"final_merged_df['severity'].value_counts()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 행마다 고유한 키값으로 row_id 열을 생성\nfinal_merged_df['row_id'] = (\n    final_merged_df['study_id'].astype(str) + '_' +\n    final_merged_df['condition'].str.lower().str.replace(' ', '_') + '_' +\n    final_merged_df['level'].str.lower().str.replace('/', '_')\n)\n\n# 행마다 그에 해당하는 이미지 경로를 담은 열을 생성\nfinal_merged_df['image_path'] = (\n    f'{train_path}/train_images/' + \n    final_merged_df['study_id'].astype(str) + '/' +\n    final_merged_df['series_id'].astype(str) + '/' +\n    final_merged_df['instance_number'].astype(str) + '.dcm'\n)\n\n# 심각도 레이블을 다루기 쉽게 변경\nfinal_merged_df['severity'] = final_merged_df['severity'].map({'Normal/Mild': 'normal_mild', 'Moderate': 'moderate', 'Severe': 'severe'})\n\n# Note: Check image path, since there's 1 instance id, for 1 image, but there's many more images other than the ones labelled in the instance ID. \n\n# Display the updated dataframe\nfinal_merged_df.head(5)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"Test Description 데이터 프레임의 전처리","metadata":{}},{"cell_type":"code","source":"# test 이미지가 포함된 series의 촬영정보가 담긴 test_desc\n# series_description으로부터 예측가능한 condition에 대한 정보를 연결하고\n# 각 condition에 대해 test_images에 담긴 이미지 파일들을 연결해야 함\ntest_desc\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# test 이미지 파일이 담긴 폴더의 경로\nbase_path = os.path.join(train_path, 'test_images')\n\n# 읽어들인 행에서 'study_id', 'series_id' 컬럼값을 불러와 이미지 파일이 있는 series 폴더 경로를 생성\n# 해당 경로가 존재하는 경우, 폴더 내에 존재하는 파일들의 경로를 리스트 형태로 반환\ndef get_image_paths(row):\n    series_folder_path = os.path.join(base_path, str(row['study_id']), str(row['series_id']))\n    if os.path.exists(series_folder_path):\n        return [os.path.join(series_folder_path, f) for f in os.listdir(series_folder_path) if f.endswith('.dcm')]\n    else:\n        print(f\"Series path does not exist: {series_folder_path}\")  # 이미지 경로가 존재하지 않는 경우 출력\n    return []\n\n# series_description으로부터 condition을 할당\n# 'side'에 좌우구분이 있는 경우('left', 'right')와 없는 경우('both')를 구분\ncondition_df = pd.DataFrame([\n    {'series_description': 'Sagittal T1', 'condition': 'left_neural_foraminal_narrowing', 'side': 'left'},\n    {'series_description': 'Sagittal T1', 'condition': 'right_neural_foraminal_narrowing', 'side': 'right'},\n    {'series_description': 'Axial T2', 'condition': 'left_subarticular_stenosis', 'side': 'left'},\n    {'series_description': 'Axial T2', 'condition': 'right_subarticular_stenosis', 'side': 'right'},\n    {'series_description': 'Sagittal T2/STIR', 'condition': 'spinal_canal_stenosis', 'side': 'both'}\n])\n\nmerged_df = pd.merge(test_desc, condition_df, on='series_description', how='left')\n\n# 'image_paths' 컬럼을 생성해, 해당 행의 'study_id' 및 'series_id'에 속한 모든 이미지 경로의 리스트를 추가\nmerged_df['image_paths'] = merged_df.apply(get_image_paths, axis=1)\n\n# explode 메서드를 통해 하나의 행에 저장된 해당 폴더의 전체 이미지 경로를 각 행으로 확장하고 다듬기\nexpanded_test_desc = merged_df.explode('image_paths') # 이미지 경로 리스트를 각 이미지 경로로 확장하기\nexpanded_test_desc = expanded_test_desc.rename(columns={'image_paths': 'image_path'}) # 'image_paths' 컬럼의 이름을 'image_path'로 변경\nexpanded_test_desc['row_id'] = expanded_test_desc.apply(lambda row: f\"{row['study_id']}_{row['condition']}\", axis=1) # 새로운 'row_id' 컬럼 추가\nexpanded_test_desc = expanded_test_desc.drop(columns=['side']) # 'side' 컬럼 드롭\nexpanded_test_desc  = expanded_test_desc.reset_index(drop=True) # 인덱스 재설정\n\n# 결과 출력\nexpanded_test_desc.head(5)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"expanded_test_desc[['series_description', 'condition']].value_counts()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"final_merged_df","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data = final_merged_df\ntest_data = expanded_test_desc\n\nprint(f\"\\ntrain data shape: {train_data.shape}\")\nprint(', '.join(train_data.columns))\nprint(f\"\\ntest data shape: {test_data.shape}\")\nprint(', '.join(train_data.columns.intersection(test_data.columns))) #train_data.columns 순서대로 정렬","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# train_data 데이터프레임에 포함된 파일 경로가 실제로 존재하는지 확인\n\n# 주어진 path가 실제 파일 시스템에 존재하는지 확인하는 함수\ndef check_exists(path):\n    return os.path.exists(path)\n\n# 각 row에 대해 study_id에 해당하는 디렉토리가 존재하는지 확인\ndef check_study_id(row):\n    study_id = row['study_id']\n    path = f'{train_path}/train_images/{study_id}'\n    return check_exists(path)\n\n# 각 row에 대해 series_id에 해당하는 디렉토리가 존재하는지 확인\ndef check_series_id(row):\n    study_id = row['study_id']\n    series_id = row['series_id']\n    path = f'{train_path}/train_images/{study_id}/{series_id}'\n    return check_exists(path)\n\n# 각 row에 대해 image_path에 해당하는 파일이 존재하는지 확인\ndef check_image_exists(row):\n    image_path = row['image_path']\n    return check_exists(image_path)\n\n# 각 계층의 디렉토리/파일의 존재여부를 확인하여 새로운 컬럼 'study_id_exists', 'series_id_exists', 'image_exists'에 추가\ntrain_data['study_id_exists'] = train_data.apply(check_study_id, axis=1)\ntrain_data['series_id_exists'] = train_data.apply(check_series_id, axis=1)\ntrain_data['image_exists'] = train_data.apply(check_image_exists, axis=1)\n\n# study_id_exists, series_id_exists, image_exists 컬럼이 모두 True인 행만 남김\ntrain_data = train_data[(train_data['study_id_exists']) & (train_data['series_id_exists']) & (train_data['image_exists'])]\n\nprint(f\"train data shape: {train_data.shape}\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_data.head(3)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 이미지 파일을 읽어들인 뒤 MinMax scaling\ndef load_dicom(path):\n    ds = dicom.dcmread(path)\n    img = ds.pixel_array\n    img = (img - np.min(img)) / (np.max(img) - np.min(img) + 1e-8)\n    img = (img * 255).astype(np.uint8)\n    return img\n\n# Load images randomly\nimport random\nimages = []\nrow_ids = []\nselected_indices = random.sample(range(len(train_data)), 2)\nfor i in selected_indices:\n    image = load_dicom(train_data['image_path'][i])\n    images.append(image)\n    row_ids.append(train_data['row_id'][i])\n\n# Plot images\nfig, ax = plt.subplots(1, 2, figsize=(8, 4))\nfor i in range(2):\n    ax[i].imshow(images[i], cmap=plt.cm.bone)\n    ax[i].set_title(f'Row ID: {row_ids[i]}', fontsize=8)\n    ax[i].axis('off')\nplt.tight_layout()\nplt.show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 결측치 처리\nprint(train_data.isna().sum(), \"\\n\")\nprint(pd.crosstab(train_data['condition'], train_data['severity'].isna()))\ntrain_data = train_data.dropna()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Making Custom Dataset","metadata":{}},{"cell_type":"code","source":"import torch\nimport torch.optim.lr_scheduler as lr_scheduler\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as transforms\nfrom sklearn.model_selection import train_test_split\nfrom tqdm import tqdm","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Define a custom dataset class\n# multi-label multi-class problem으로 정의\n# 레벨이 L1/L2이고, 심각도가 severe라면 [2, 0, 0, 0, 0]\n# 레벨이 L3/L4이고, 심각도가 moderate라면 [0, 0, 1, 0, 0]\nclass CustomDataset(Dataset):\n    def __init__(self, dataframe, transform=None):\n        self.dataframe = dataframe\n        self.transform = transform\n        self.level_map = {'L1/L2': 0, 'L2/L3': 1, 'L3/L4': 2, 'L4/L5': 3, 'L5/S1': 4}  # 레벨 인덱스 맵\n        self.severity_map = {'normal_mild': 0, 'moderate': 1, 'severe': 2}  # 심각도 맵핑\n\n    def __len__(self):\n        return len(self.dataframe)\n\n    def __getitem__(self, index):\n        image_path = self.dataframe['image_path'][index]\n        image = load_dicom(image_path)  # DICOM 이미지를 로드하는 함수\n        level = self.dataframe['level'][index]  # 레벨 정보 추출\n        severity = self.dataframe['severity'][index]  # 심각도 정보 추출\n        \n        label = np.zeros(5, dtype=int)  # 5개의 원소를 가지는 ndarray 초기화\n        label[self.level_map[level]] = self.severity_map[severity]  # 레벨 인덱스에 심각도 매핑\n        \n        if self.transform:\n            image = self.transform(image)  # 필요한 경우 이미지 변환 적용\n\n        # label을 torch 텐서로 변환\n        label = torch.tensor(label, dtype=torch.long)\n        \n        return image, label","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Function to create datasets and dataloaders for each series description\ndef create_datasets_and_loaders(df, series_description, transform, batch_size=8):\n    #series_description에 맞는 행만 필터링하여 filtered_df에 저장\n    filtered_df = df[df['series_description'] == series_description]\n    \n    #filtered_df를 0.8/0.2로 분할하여 train_df, val_df에 저장한 뒤 인덱스 초기화\n    train_df, val_df = train_test_split(filtered_df, test_size=0.2, random_state=42)\n    train_df = train_df.reset_index(drop=True)\n    val_df = val_df.reset_index(drop=True)\n\n    #train_df, val_df와 transform을 결합하여 CustomDataset 객체 train_dataset, val_dataset 생성\n    train_dataset = CustomDataset(train_df, transform)\n    val_dataset = CustomDataset(val_df, transform)\n\n    #인수로 받은 batch_size를 적용하여 dataset을 미니배치로 분할하여 DataLoader 객체 loader에 저장\n    trainloader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)\n    valloader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)\n    \n    return trainloader, valloader, len(train_df), len(val_df)\n\n# Define the transforms\ntransform = transforms.Compose([\n    #DICOM 파일은 0~1 사이의 실수로 구성되는데, PIL 이미지로 변환하기 위해 0~255 사이 정수로 변환\n    transforms.Lambda(lambda x: (x * 255).astype(np.uint8)), \n    #PIL(Python Imaging Library) 이미지로 변환 \n    transforms.ToPILImage(),\n    #이미지를 (224, 224) 크기로 조정 - 사전 훈련된 모델이 이미지를 받는 크기\n    transforms.Resize((224, 224)),\n    #이미지를 그레이스케일로 변환하고, 3개의 채널(RGB)로 복제 - 사전 훈련된 모델이 이미지를 받는 형식\n    transforms.Grayscale(num_output_channels=3),\n    #PIL 이미지를 PyTorch 텐서로 변환\n    transforms.ToTensor(),\n])\n\n# Create dataloaders for each series description\ndataloaders = {}\nlengths = {}\n\n#train_data에서 series_description이 'xx'인 행을 추출하고, train/valid set을 분할하여 CustomDataset 객체로 만들고\n#batch_size=8로 분할하여 trainloader_xx, valloader_xx를 구성하고, 길이 len_train_t1, len_val_t1와 함께 반환\ntrainloader_t1, valloader_t1, len_train_t1, len_val_t1 = create_datasets_and_loaders(train_data, 'Sagittal T1', transform)\ntrainloader_t2, valloader_t2, len_train_t2, len_val_t2 = create_datasets_and_loaders(train_data, 'Axial T2', transform)\ntrainloader_t2stir, valloader_t2stir, len_train_t2stir, len_val_t2stir = create_datasets_and_loaders(train_data, 'Sagittal T2/STIR', transform)\n\n#dataloaders 딕셔너리에 각 key마다 series_description을 할당하고, (trainloader, valloader)를 저장\ndataloaders['Sagittal T1'] = (trainloader_t1, valloader_t1)\ndataloaders['Axial T2'] = (trainloader_t2, valloader_t2)\ndataloaders['Sagittal T2/STIR'] = (trainloader_t2stir, valloader_t2stir)\n\n#lengths 딕셔너리에 각 key마다 series_description을 할당하고, (len_train,len_val)을 저장\nlengths['Sagittal T1'] = (len_train_t1, len_val_t1)\nlengths['Axial T2'] = (len_train_t2, len_val_t2)\nlengths['Sagittal T2/STIR'] = (len_train_t2stir, len_val_t2stir)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Function to visualize a batch of images\ndef visualize_batch(dataloader):\n    images, labels = next(iter(dataloader))\n    fig, axes = plt.subplots(1, len(images), figsize=(20, 5))\n    for i, (img, lbl) in enumerate(zip(images, labels)):\n        ax = axes[i]\n        img = img.permute(1, 2, 0)  # Convert to HWC for visualization\n        ax.imshow(img)\n        ax.set_title(f\"Label: {lbl}\")\n        ax.axis('off')\n    plt.show()\n\n# Visualize samples from each dataloader\nprint(\"Visualizing Sagittal T1 samples\")\nvisualize_batch(trainloader_t1)\nprint(\"Visualizing Axial T2 samples\")\nvisualize_batch(trainloader_t2)\nprint(\"Visualizing Sagittal T2/STIR samples\")\nvisualize_batch(trainloader_t2stir)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"image, label = next(iter(trainloader_t1))\nsample = image[1].permute(1, 2, 0)  #sample\nplt.imshow(sample)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Plot images\nimage, label = next(iter(trainloader_t2))\nplt.figsize=(8, 4)\nplt.imshow(images[0], cmap='gray')\nplt.title(label[0])\nplt.axis('off')\nplt.tight_layout()\nplt.show()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Model","metadata":{}},{"cell_type":"code","source":"import torch\nimport torch.nn as nn\nimport torchvision.models as models\nfrom torchvision import transforms\nfrom torch.utils.data import DataLoader\nfrom sklearn.model_selection import train_test_split\nimport pandas as pd\nfrom tqdm import tqdm","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#PyTorch의 nn.Module을 상속받아 사용자 정의 EfficientNetV2 모델을 구현\nclass CustomEfficientNetV2(nn.Module):\n    #num_classes=3: 분류할 클래스의 수를 지정하는 파라미터\n    #pretrained_weights=None: 사전 학습된 가중치 파일의 경로\n    def __init__(self, num_classes=3,pretrained_weights=None):\n        #EfficientNetV2 모델을 불러와 self.model을 초기화\n        #부모 클래스인 nn.Module의 초기화 메소드를 호출. nn.Module 클래스의 메소드와 속성들을 사용할 수 있게 됨.\n        super(CustomEfficientNetV2, self).__init__() \n        self.model = models.efficientnet_v2_s(weights=None) #torchvision.models 모듈에서 efficientnet_v2_s 모델을 불러와 self.model에 할당\n        #self.model = models.efficientnet_v2_s(weights=models.EfficientNet_V2_S_Weights.IMAGENET1K_V1)\n        #arg로 받은 파라미터가 있다면 로딩\n        if pretrained_weights:\n            self.model.load_state_dict(torch.load(pretrained_weights, weights_only=True))\n        #모델의 classifier 부분 마지막 층의 출력을 num_classes로 설정\n        #마지막 층 : self.model.classifier[-1] \n        #마지막 층 : nn.Linear(in_features, out_features)\n        num_ftrs = self.model.classifier[-1].in_features #마지막 층의 입력 피쳐수를 받아\n        self.model.classifier[-1] = nn.Linear(num_ftrs, num_classes * 5) #5개의 레벨에 대해 각각 3개의 클래스 예측\n        \n    #모델의 순전파를 정의. x를 모델에 통과시켜 예측값을 반환. 모델을 호출할 때 자동으로 실행됨.\n    def forward(self, x):\n        x = self.model(x)\n        x = x.view(-1, 5, 3)  # 출력 크기를 (배치 크기, 5개의 레벨, 3개의 클래스)로 조정\n        return x\n\n    #학습가능한 레이어 설정(별도의 코드로 전체 패러미터를 동결한 후에 실행)\n    def unfreeze_model(self, num_unfreeze=2):\n        # 특징추출기(self.model.features)의 마지막 num_unfreeze개층 중에서 BatchNorm을 제외한 층\n        for layer in list(self.model.features.children())[-num_unfreeze:]: #children() 메서드는 features의 하위 모듈(레이어)을 하나씩 반환하는 iterator 제공\n            if not isinstance(layer, nn.BatchNorm2d):\n                for param in layer.parameters():\n                    param.requires_grad = True\n        \n        # 분류기(self.model.classifier)에 해당하는 전체 층\n        for param in self.model.classifier.parameters():\n            param.requires_grad = True","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 로컬에서 작동할 때에만 설정\n#weights = models.EfficientNet_V2_S_Weights.IMAGENET1K_V1 # EfficientNet V2 S 모델의 가중치 객체 가져오기\n#url = weights.url # 가중치의 URL 가져오기\n#print(f\"URL for the pretrained weights: {url}\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#GPU 가속설정\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n#device = torch.device(\"mps\" if torch.cuda.is_available() else \"cpu\")\n\n# Path to the locally uploaded weights file\n#weights_path = train_path + '/params/efficientnet_v2_s-dd5fe13b.pth'\nweights_path = '/kaggle/input/efficientnetv2/pytorch/efficientnetv2s/1/efficientnet_v2_s-dd5fe13b.pth'\n\n# Initialize models\nsagittal_t1_model = CustomEfficientNetV2(num_classes=3, pretrained_weights=weights_path).to(device)\naxial_t2_model = CustomEfficientNetV2(num_classes=3, pretrained_weights=weights_path).to(device)\nsagittal_t2stir_model = CustomEfficientNetV2(num_classes=3, pretrained_weights=weights_path).to(device)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# features 내부의 각 모듈을 재귀적으로 탐색하여 모든 층을 출력\nnum_f_layers = len(list(sagittal_t1_model.model.features.children()))\nprint(f\"The number of layers in self.model.features: {num_f_layers}\")\n\nnum_f_layers = len(list(sagittal_t1_model.model.classifier.children()))\nprint(f\"The number of layers in self.model.features: {num_f_layers}\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 모델의 특징추출기를 동결\nfor param in sagittal_t1_model.model.features.parameters():\n    param.requires_grad = False\nfor param in axial_t2_model.model.features.parameters():\n    param.requires_grad = False\nfor param in sagittal_t2stir_model.model.features.parameters():\n    param.requires_grad = False\n\n# 일부 층만 다시 학습가능하도록 설정\n#sagittal_t1_model.unfreeze_model(num_unfreeze=1)\n#axial_t2_model.unfreeze_model(num_unfreeze=1)\n#sagittal_t2stir_model.unfreeze_model(num_unfreeze=1)\n\n# 모델의 분류기를 학습가능하도록 설정\nfor param in sagittal_t1_model.model.classifier.parameters():\n    param.requires_grad = True\nfor param in axial_t2_model.model.classifier.parameters():\n    param.requires_grad = True\nfor param in sagittal_t2stir_model.model.classifier.parameters():\n    param.requires_grad = True\n\n#손실함수를 정의\nweights = torch.tensor([1.0, 2.0, 4.0]).to(device) # 가중치 정의\ndef weighted_loss_function(outputs, labels):\n    loss = 0\n    for i in range(5):\n        loss_fn = nn.CrossEntropyLoss(weight=weights)  # 가중치가 적용된 CrossEntropyLoss\n        loss += loss_fn(outputs[:, i, :], labels[:, i])  # 각 레벨별 손실 계산 및 가중치 적용\n    return loss\n\n# 학습을 위한 optimizer 초기화\noptimizer_sagittal_t1 = torch.optim.Adam(sagittal_t1_model.model.classifier.parameters(), lr=0.001)\noptimizer_axial_t2 = torch.optim.Adam(axial_t2_model.model.classifier.parameters(), lr=0.001)\noptimizer_sagittal_t2stir = torch.optim.Adam(sagittal_t2stir_model.model.classifier.parameters(), lr=0.001)\n\n# 빠른 액세스를 위해 모델과 optimizer를 딕셔너리로 저장\ncustom_models = {\n    'Sagittal T1': sagittal_t1_model,\n    'Axial T2': axial_t2_model,\n    'Sagittal T2/STIR': sagittal_t2stir_model,\n}\noptimizers = {\n    'Sagittal T1': optimizer_sagittal_t1,\n    'Axial T2': optimizer_axial_t2,\n    'Sagittal T2/STIR': optimizer_sagittal_t2stir,\n}","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Count trainable parameters\ntrainable_params = sum(p.numel() for p in sagittal_t1_model.parameters() if p.requires_grad)\nprint(f\"Number of parameters: {trainable_params}\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def print_pytorch_model_summary(model, input_size=(3, 224, 224)):\n    print(f\"{'Layer Name':<30} {'Output Shape':<30} {'Param #':<15} {'Type':<15}\")\n    print(\"=\"*100)\n    \n    # Create a dummy input tensor with the specified input size\n    dummy_input = torch.rand(1, *input_size).to(next(model.parameters()).device)\n    \n    # Forward pass through the model to get output shapes\n    def register_hook(module, module_type):\n        def hook(module, input, output):\n            class_name = str(module.__class__).split(\".\")[-1].split(\"'\")[0]\n            module_idx = len(summary)\n\n            # Module name and output shape\n            m_key = f\"{class_name}-{module_idx + 1}\"\n            summary[m_key] = {\n                \"input_shape\": list(input[0].size()),\n                \"output_shape\": list(output.size()),\n                \"nb_params\": sum(p.numel() for p in module.parameters()),\n                \"type\": module_type\n            }\n        \n        if (\n            not isinstance(module, nn.Sequential) and \n            not isinstance(module, nn.ModuleList) and \n            module != model\n        ):\n            hooks.append(module.register_forward_hook(hook))\n    \n    # Register hooks for features and classifier separately\n    summary = {}\n    hooks = []\n    \n    # Register hooks for feature extraction layers\n    model.model.features.apply(lambda module: register_hook(module, \"Feature Extraction\"))\n    \n    # Register hooks for classifier layers\n    model.model.classifier.apply(lambda module: register_hook(module, \"Classifier\"))\n\n    # Perform a forward pass to collect output shapes\n    model(dummy_input)\n    \n    # Remove hooks\n    for hook in hooks:\n        hook.remove()\n\n    total_params = 0\n    for layer in summary:\n        # Layer name\n        print(f\"{layer:<30} {str(summary[layer]['output_shape']):<30} {summary[layer]['nb_params']:<15} {summary[layer]['type']:<15}\")\n        total_params += summary[layer][\"nb_params\"]\n    \n    print(\"=\"*100)\n    print(f\"Total Params: {total_params}\")\n\n# 모델 서머리 출력 (예: sagittal_t1_model)\n#print_pytorch_model_summary(sagittal_t1_model)\n\n# features 내부의 각 모듈을 재귀적으로 탐색하여 모든 층을 출력\ndef count_layers(module):\n    num_layers = 0\n    for child in module.children():\n        if isinstance(child, nn.Sequential):\n            num_layers += count_layers(child)\n        else:\n            num_layers += 1\n    return num_layers\n\nnum_f_layers = count_layers(sagittal_t1_model.model.features)\nprint(f\"The number of layers in self.model.features: {num_f_layers}\")\n\nnum_c_layers = count_layers(sagittal_t1_model.model.classifier)\nprint(f\"The number of layers in self.model.classifier: {num_c_layers}\")\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Training","metadata":{}},{"cell_type":"code","source":"# trainloader에서 불러온 입출력이 device로 정상적으로 전달되는지 체크\nfor images, labels in trainloader_t2:\n    labels = labels.clone().detach().to(device)  # 이미 텐서인 경우 clone().detach()를 사용\n    labels = labels.to(device)  # 데이터를 GPU device로 할당\n    \n    print(labels)\n    break","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import torch.optim.lr_scheduler as lr_scheduler\nfrom copy import deepcopy\n\n#학습절차 정의\ndef train_model(model, trainloader, valloader, len_train, len_val, optimizer, num_epochs=20, patience=3):\n    # StepLR: 단계적 학습률 감소. 2 epoch마다 학습률이 0.1 감소.\n    scheduler = lr_scheduler.StepLR(optimizer, step_size=2, gamma=0.1)\n    \n    #from pytorch_lightning.callbacks import EarlyStopping\n    #early_stop_callback = EarlyStopping(monitor='val_loss', patience, verbose=True)\n    #trainer = Trainer(callbacks=[early_stop_callback])\n    #trainer.fit(model, trainloader, valloader)\n\n    # 조기 종료를 위해 사용하는 임시변수 초기화\n    best_val_acc = 0.0  # 검증 정확도의 최고 값을 저장할 변수\n    best_model_wts = deepcopy(model.state_dict())  # 모델의 가중치를 저장하기 위한 변수\n    counter = 0  # 조기 종료를 위한 에포크 카운터\n    \n    for epoch in range(num_epochs): #총 num_epochs만큼 반복\n        model.train() #모델을 학습 모드로 설정\n        train_loss = 0\n        correct_train = 0\n        \n        with tqdm(trainloader, unit=\"batch\") as tepoch: #train 데이터를 tqdm으로 감싸고 배치 단위로 반복\n            for images, labels in tepoch: \n                images, labels = images.to(device), labels.clone().detach().to(device) #데이터를 GPU device로 이동\n                optimizer.zero_grad() #옵티마이저의 기울기 초기화\n                outputs = model(images) #모델에 입력 이미지를 넣어 출력 계산\n                loss = weighted_loss_function(outputs, labels) #출력과 실제 레이블 간의 손실 계산\n                loss.backward() #역전파를 통해 기울기 계산\n                optimizer.step() #옵티마이저를 통해 가중치 업데이트\n                train_loss += loss.item() #배치의 손실을 train_loss에 더함\n\n                # 레벨별로 예측 결과와 실제 레이블을 비교\n                for i in range(5):  # 5개의 레벨에 대해 반복\n                    probabilities = torch.softmax(outputs[:, i, :], dim=1)\n                    _, predicted = torch.max(probabilities, 1)\n                    correct_train += (predicted == labels[:, i]).sum().item()  # 해당 레벨의 정확도를 계산\n                \n                tepoch.set_postfix(epoch=epoch+1)\n                \n        scheduler.step() #학습률 스케줄러의 스텝을 증가시킴\n        \n        #train set의 loss, accuracy 계산\n        train_loss /= len(trainloader) #average loss(train set)\n        train_acc = 100*(correct_train/len_train)/5 #accuracy(train set)\n        \n        model.eval() #모델을 평가 모드로 설정\n        val_loss, correct_val = 0, 0 #검증 손실 및 맞춘 검증 데이터 개수 초기화\n        with torch.no_grad(): #기울기 계산 비활성화(평가에서는 불필요)\n            with tqdm(valloader, unit=\"batch\") as vepoch: #valid 데이터를 tqdm으로 감싸고 배치 단위로 반복\n                for images, labels in vepoch:\n                    images, labels = images.to(device), labels.clone().detach().to(device) #데이터를 GPU device로 이동\n                    outputs = model(images) #모델에 입력 이미지를 넣어 출력 계산\n                    loss = weighted_loss_function(outputs, labels) #출력과 실제 레이블 간의 손실 계산\n                    val_loss += loss.item() #배치의 손실을 val_loss에 더함\n\n                    for i in range(5):\n                        probabilities = torch.softmax(outputs[:, i, :], dim=1)\n                        _, predicted = torch.max(probabilities, 1)\n                        correct_val += (predicted == labels[:, i]).sum().item()\n                    \n                    vepoch.set_postfix(epoch=epoch+1) #tqdm 진행바에 현재 에포크를 출력\n        \n        #valid set의 loss, accuracy 계산\n        val_loss /= len(valloader)  #average loss\n        val_acc = 100*(correct_val/len_val)/5 #accuracy\n        \n        #현재 에포크의 학습 및 검증 결과 출력\n        print(f\"Epoch {epoch+1}, Train Loss: {train_loss:.4f}, Train Acc: {train_acc:.2f}%, Val Loss: {val_loss:.4f}, Val Acc: {val_acc:.2f}%\")\n        \n        #최고 accuracy를 가진 모델 저장\n        if val_acc > best_val_acc:\n            best_val_acc = val_acc #최고 accuracy 갱신\n            best_model_wts = deepcopy(model.state_dict()) #최적의 모델 가중치 저장\n            counter = 0 #조기 종료 카운터 초기화\n            torch.save(best_model_wts, f'best_model_{epoch+1}.pth') #모델 가중치 저장\n        else:\n            counter += 1 #현재 에포크가 best가 아닌 경우 조기 종료 카운터 증가\n        \n        #설정된 patience만큼 개선이 이루어지지 않았으면 종료\n        if counter >= patience:\n            print(f\"Early stopping triggered after {epoch+1} epochs\")\n            break\n    \n    #학습 후 최적의 가중치를 모델에 적용\n    model.load_state_dict(best_model_wts)\n    return model, best_val_acc #최적의 모델과 검증 정확도를 반환","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"'''이 부분이 병목\n#trainloader_t2에서 불러온 labels 텐서 내에 NaN(Not a Number) 값이 있는지 체크\n#NaN 값은 수학적으로 정의되지 않은 값으로, 학습 과정에서 NaN 값이 발생하면 모델 성능에 심각한 영향을 미칠 수 있음\nfor images, labels in trainloader_t2:\n    if any(label != label for label in labels):  # nan은 자기 자신과 같지 않기 때문에 이를 이용한 체크\n        print(\"Found nan label\")\n'''","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# GPU 가속이 가능한지 확인하는 코드\nif torch.cuda.is_available():\n    print(\"CUDA is available. GPU acceleration is enabled.\")\n    print(f\"Device name: {torch.cuda.get_device_name(0)}\")\nelse:\n    print(\"CUDA is not available. Running on CPU.\")","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Training all models\nfor desc, model in custom_models.items(): #key와 value를 튜플형태로 조회\n    if desc == 'Sagittal T1':\n        trainloader, valloader, len_train, len_val = trainloader_t1, valloader_t1, len_train_t1, len_val_t1\n    elif desc == 'Axial T2':\n        trainloader, valloader, len_train, len_val = trainloader_t2, valloader_t2, len_train_t2, len_val_t2\n    elif desc == 'Sagittal T2/STIR':\n        trainloader, valloader, len_train, len_val = trainloader_t2stir, valloader_t2stir, len_train_t2stir, len_val_t2stir\n    \n    print(f\"Training model for {desc}\")\n    train_model(model, trainloader, valloader, len_train, len_val, optimizers[desc])","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Inference","metadata":{}},{"cell_type":"code","source":"#train data에 존재하는 level 확인\ntrain_data['level'].unique() ","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#test data에는 사진만 있고 level 정보가 없음\nexpanded_test_desc.head(5)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"levels = ['l1_l2', 'l2_l3', 'l3_l4', 'l4_l5', 'l5_s1']\n\n#row_id에 levels('l1_l2'~'l5_s1')을 붙여서 반환하는 함수\ndef update_row_id(row, levels): #인덱스 번호를 levels 리스트의 길이로 나눈 나머지를 계산하여 추가\n    level = levels[row.name % len(levels)] #apply 함수 내에서 row.name은 현재 행의 인덱스 번호를 나타냄\n    return f\"{row['study_id']}_{row['condition']}_{level}\"\n\n#row_id에 levels를 붙여서 expanded_test_desc를 확장 \nexpanded_test_desc['row_id'] = expanded_test_desc.apply(lambda row: update_row_id(row, levels), axis=1)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"expanded_test_desc.head()","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"expanded_test_desc[expanded_test_desc['series_description'] == 'Sagittal T2/STIR']['image_path'].str.split('/').str[-3:].str.join('/')","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Define a custom test dataset class\nclass TestDataset(Dataset):\n    #instance = CustomDataset(test_data)\n    #instance.dataframe -> dataframe\n    #instance.transform -> transform\n    def __init__(self, dataframe, transform=None):\n        self.dataframe = dataframe\n        self.transform = transform\n\n    #len(instance) -> len(dataframe)\n    def __len__(self):\n        return len(self.dataframe)\n\n    #instance[index] -> image, label\n    def __getitem__(self, index):\n        image_path = self.dataframe['image_path'][index] #인덱스를 통해 경로 호출\n        image = load_dicom(image_path)  #경로로부터 DICOM image 로딩 #train과 달리 label은 없음!\n        if self.transform:\n            image = self.transform(image) #정의된 변환이 있을 경우 이미지에 변환 적용(리스케일 후 PyTorch Tensor로)\n        return image \n\n# Define the transforms\ntransform = transforms.Compose([\n    transforms.ToPILImage(),\n    transforms.Resize((224, 224)),\n    transforms.Grayscale(num_output_channels=3),\n    transforms.ToTensor(),\n])\n\n# Create a test dataset and dataloader\ntest_dataset = TestDataset(expanded_test_desc, transform) #TestDataset에 expanded_test_desc를 넣어 이미지를 불러올 준비를 마침\ntestloader = DataLoader(test_dataset, batch_size=1, shuffle=False) #batch_size=1으로 한 번에 하나의 샘플에 대해 예측을 수행DataLoader 객체 선언","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#testloader에 이미지가 정확한 형태로 로딩되었는지 확인\n#첫번째 루프(첫 번째 배치)만 불러와 형태를 출력한 뒤 break \n#(batch_size, channels, height, width)\nfor image in testloader:\n    print(image.shape)\n    break","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def get_model(series_description):\n    return custom_models.get(series_description, None)\n\ndef predict_test_data(testloader, expanded_test_desc):\n    # 예측 결과를 저장할 리스트 선언\n    predictions = []\n    \n    # 모든 모델을 평가 모드로 설정\n    for model in custom_models.values():\n        model.eval()\n\n    # 추론 과정에서는 기울기 계산을 비활성화하여 메모리 사용을 줄임   \n    with torch.no_grad():\n        # 전체 인덱스를 위한 카운터\n        total_index = 0\n        # testloader에서 배치 단위로 데이터를 가져옴\n        for images in tqdm(testloader):\n            images = images.to(device)  # 이미지를 GPU로 이동\n            batch_size = images.size(0) # 현재 배치의 크기 확인\n\n            for i in range(batch_size):\n                series_description = expanded_test_desc.iloc[total_index]['series_description']  # 현재 이미지에 해당하는 series_description\n                model = get_model(series_description)  # 해당 series description에 맞는 모델\n                if model:  # 모델이 존재하면\n                    outputs = model(images[i].unsqueeze(0))  # (1, 5, 3)의 shape\n                    probs = torch.softmax(outputs, dim=2).squeeze(0)  # 각 레벨에 대한 클래스 확률 계산 -> pytorch tensor 객체\n\n                    # 각 레벨별로 예측된 확률을 저장\n                    level_predictions = []\n                    for level_idx in range(5):\n                        level_predictions.append({\n                            'normal_mild': probs[level_idx, 0].item(),\n                            'moderate': probs[level_idx, 1].item(),\n                            'severe': probs[level_idx, 2].item(),\n                            'level': level_idx,\n                            'series_description': series_description\n                        })\n                    predictions.append(level_predictions)\n                else:\n                    # 모델이 없으면 None으로 채움\n                    predictions.append([{'normal_mild': None, 'moderate': None, 'severe': None, 'level': None, 'condition': None}] * 5)\n                \n                # 다음 이미지로 넘어가기 위해 전체 인덱스 증가\n                total_index += 1\n    \n    return predictions\n","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 예측 수행\nresults = predict_test_data(testloader, expanded_test_desc)\n\nresults_df = pd.DataFrame()\n\nfor result in results:\n    row = pd.DataFrame(result)\n    results_df = pd.concat([results_df, row])","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"condition_df = pd.DataFrame([\n    {'series_description': 'Sagittal T1', 'condition': 'left_neural_foraminal_narrowing'},\n    {'series_description': 'Sagittal T1', 'condition': 'right_neural_foraminal_narrowing'},\n    {'series_description': 'Axial T2', 'condition': 'left_subarticular_stenosis'},\n    {'series_description': 'Axial T2', 'condition': 'right_subarticular_stenosis'},\n    {'series_description': 'Sagittal T2/STIR', 'condition': 'spinal_canal_stenosis'}\n])\n\nresults_df_with_condition = pd.merge(results_df, condition_df, on='series_description', how='left')\n\ninverse_level_map = {0 : 'l1_l2', 1:'l2_l3', 2: 'l3_l4', 3: 'l4_l5', 4: 'l5_s1'}\nresults_df_with_condition['level'] = results_df_with_condition['level'].map(inverse_level_map)\nresults_df_with_condition['row_id'] = \"44036939_\" + results_df_with_condition['condition'] + \"_\" + results_df_with_condition['level']\nresults_df_with_condition = results_df_with_condition[['row_id', 'normal_mild', 'moderate', 'severe']]","metadata":{"execution":{"iopub.status.busy":"2024-08-29T16:26:36.082491Z","iopub.execute_input":"2024-08-29T16:26:36.082967Z","iopub.status.idle":"2024-08-29T16:26:36.529644Z","shell.execute_reply.started":"2024-08-29T16:26:36.082930Z","shell.execute_reply":"2024-08-29T16:26:36.527858Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# 'row_id'를 기준으로 그룹화하고 각 레벨의 평균을 계산\nsubmission = results_df_with_condition.groupby('row_id').mean().reset_index()\n\n# 합계가 1이 되는지 확인\nrow_sum = submission[['normal_mild', 'moderate', 'severe']].sum(axis=1)\n\n# 제출 파일 저장\nsubmission.to_csv(\"/kaggle/working/submission.csv\", index=False)","metadata":{},"execution_count":null,"outputs":[]}]}