{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":52254,"databundleVersionId":9674523,"sourceType":"competition"}],"dockerImageVersionId":31234,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"!pip install monai","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport numpy as np\nimport pandas as pd\nimport pydicom\nimport torch\nfrom tqdm.auto import tqdm\nfrom monai.transforms import (\n    Compose, LoadImaged, Spacingd, Orientationd, EnsureChannelFirstd,\n    ScaleIntensityRanged, Resized, CopyItemsd, ConcatItemsd, DeleteItemsd, MapTransform,\n    CropForegroundd\n)\nfrom monai.data import MetaTensor\nimport shutil\n\nclass SelectiveSamplingd(MapTransform):\n    def __init__(self, keys, num_slices=64):\n        super().__init__(keys)\n        self.num_slices = num_slices\n\n    def __call__(self, data):\n        d = dict(data)\n        for key in self.keys:\n            img = d[key] # (C, S, H, W)\n            s = img.shape[1]\n            # 64장을 뽑을 인덱스 계산\n            indices = np.linspace(0, s - 1, self.num_slices).astype(int)\n            d[key] = img[:, indices, :, :]\n        return d\n\nclass Preprocessor:\n    def __init__(self, save_dir, target_slices=64, target_size=160):\n        self.save_dir = save_dir\n        self.transforms = Compose([\n\n            # 1. 방향 통일 (코드 B의 장점: 해부학적 위치 고정)\n            Orientationd(keys=[\"image\"], axcodes=\"RAS\"),\n            \n            # 2. 물리적 간격 표준화 (1.5mm Isotropic)\n            Spacingd(keys=[\"image\"], pixdim=(1.5, 1.5, 1.5), mode=\"bilinear\"),\n            \n            # 3. 외상 특화 윈도잉 3채널 생성 (코드 A의 장점)\n            CopyItemsd(keys=[\"image\"], times=3, names=[\"img_soft\", \"img_angio\", \"img_bowel\"]),\n            \n            # Soft Tissue (장기 손상용)\n            ScaleIntensityRanged(keys=[\"img_soft\"], a_min=-160, a_max=240, b_min=0.0, b_max=1.0, clip=True),\n            # Angio/Blood (활성 출혈 점 강조용)\n            ScaleIntensityRanged(keys=[\"img_angio\"], a_min=-250, a_max=450, b_min=0.0, b_max=1.0, clip=True),\n            # Bowel/Air (장 천공 가스 강조용)\n            ScaleIntensityRanged(keys=[\"img_bowel\"], a_min=-300, a_max=200, b_min=0.0, b_max=1.0, clip=True),\n            \n            # 4. 채널 결합 및 불필요 항목 삭제\n            ConcatItemsd(keys=[\"img_soft\", \"img_angio\", \"img_bowel\"], name=\"image\"),\n            DeleteItemsd(keys=[\"img_soft\", \"img_angio\", \"img_bowel\"]),\n            \n            # 5. 크기 표준화 (깊이 64, 가로세로 128로 리사이징)\n            # 검은 공기(배경) 잘라내기 (환자 몸통만 남김)\n            # 윈도잉으로 공기를 0으로 만든 직후에 써야 합니다.\n            # margin=5를 주어 피부 바깥쪽 정보가 너무 칼같이 잘리지 않게 보호합니다.\n            CropForegroundd(keys=[\"image\"], source_key=\"image\", margin=5),\n            \n            # 64장 샘플링 (위의 SelectiveSamplingd 사용)\n            SelectiveSamplingd(keys=[\"image\"], num_slices=64),\n            \n            # 5. 가로세로만 리사이즈 (-1은 깊이(64장)를 유지하라는 뜻)\n            Resized(keys=[\"image\"], spatial_size=(-1, target_size, target_size))\n        ])\n\n    def get_valid_dicom_files(self, dcm_dir):\n        \"\"\"코드 A의 장점: 손상된 DICOM 파일을 사전에 걸러냄\"\"\"\n        valid_slices = []\n\n        for f in os.listdir(dcm_dir):\n            path = os.path.join(dcm_dir, f)\n            try:\n                ds = pydicom.dcmread(path)\n                _ = ds.pixel_array # RLE 디코딩 테스트\n                \n                if \"ImagePositionPatient\" in ds:\n                    valid_slices.append(ds)\n            except:\n                continue\n                \n        if len(valid_slices) < 10:\n            return None\n        \n        # Z-Coordinate 기준 정렬 (해부학적 순서)\n        valid_slices.sort(key=lambda x: float(x.ImagePositionPatient[2]))\n\n        # HU 변환 및 볼륨 생성\n        vol = []\n        for s in valid_slices:\n            img = s.pixel_array.astype(np.float32)\n            slope = valid_slices[0].RescaleSlope\n            intercept = valid_slices[0].RescaleIntercept\n            vol.append(img * slope + intercept)\n            \n        vol = np.stack(vol, axis=0)\n        \n        if len(valid_slices) > 1:\n            z_spacing = np.abs(valid_slices[1].ImagePositionPatient[2] - valid_slices[0].ImagePositionPatient[2]) \n        else:\n            z_spacing = valid_slices[0].SliceThickness\n            \n        curr_spacing = (\n            z_spacing, \n            float(valid_slices[0].PixelSpacing[0]), \n            float(valid_slices[0].PixelSpacing[1])\n        )\n        \n        # 2. 현재 Spacing을 기반으로 Affine 행렬(4x4) 생성\n        # 1.0은 monai에서 필요한 4x4를 맞추기 위해, 원래는 3x3\n        affine = np.diag(list(curr_spacing) + [1.0])\n\n        # 3. MetaTensor 생성 (데이터 + Affine을 하나로 묶음)\n        # vol: (D, H, W) -> [None] 추가하여 (C, D, H, W) 형태로 변환\n        vol_meta = MetaTensor(vol[None], affine=affine)\n        \n        return vol_meta\n\n    def run(self, data_list):\n        \n        for item in tqdm(data_list, desc=\"전처리 진행 중\"):\n            patient_id = str(item['PatientID'])\n            series_path = item['series_path']\n            series_id = series_path.split('/')[-1]\n            \n            full_dcm_path = os.path.join(BASE_DIR, series_path)\n            \n            vol_meta = self.get_valid_dicom_files(full_dcm_path)\n                \n            try:\n                # MONAI 파이프라인 실행\n                # LoadImaged는 파일 리스트를 직접 받을 수 있습니다.\n                data = self.transforms({\"image\": vol_meta})\n                \n                # 결과물 추출 (C, D, H, W) -> (3, 64, 128, 128)\n                final_volume = data[\"image\"].numpy()\n                \n                # 저장 (환자 ID별 폴더 생성)\n                out_path = os.path.join(self.save_dir, f\"{series_id}.npy\")\n                \n                # 용량 절약을 위해 float16 저장\n                np.save(out_path, final_volume.astype(np.float16))\n                \n            except Exception as e:\n                print(f\"Error processing {series_path}: {e}\")","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"BASE_DIR = '/kaggle/input/rsna-2023-abdominal-trauma-detection/'\n\n# 전처리된 데이터를 저장할 폴더\nSAVE_DIR = '/kaggle/working/processed_files'\nos.makedirs(SAVE_DIR, exist_ok=True)\n\n# 파일 읽기\ntags_df = pd.read_parquet(f'{BASE_DIR}test_dicom_tags.parquet')\n\ninjury_columns = [\n    'bowel_injury', \n    'extravasation_injury', \n    'kidney_low', 'kidney_high', \n    'liver_low', 'liver_high', \n    'spleen_low', 'spleen_high'\n]\n\n# 고유 폴더 경로 추출 및 환자 ID 연결\ntags_df['series_path'] = tags_df['path'].str.split('/').str[:-1].str.join('/')\nunique_series = tags_df[['PatientID', 'series_path']].drop_duplicates()\n\nunique_series = unique_series[:]\nunique_series_list = unique_series.to_dict('records') # 이 줄을 추가하세요\ntotal_len = len(unique_series)\n\npreprocessor = Preprocessor(SAVE_DIR)\npreprocessor.run(unique_series_list)\n \nprint(\"최종 압축 시작...\")\n# processed_files 폴더만 압축하여 'results.zip' 생성\nshutil.make_archive('/kaggle/working/final_output', 'zip', SAVE_DIR)\nprint(\"압축 완료!\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-01-16T15:03:12.432206Z","iopub.execute_input":"2026-01-16T15:03:12.436736Z","iopub.status.idle":"2026-01-16T15:03:12.817168Z","shell.execute_reply.started":"2026-01-16T15:03:12.436675Z","shell.execute_reply":"2026-01-16T15:03:12.815554Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}