{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.11","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":12249891,"sourceType":"datasetVersion","datasetId":7718540},{"sourceId":480006,"sourceType":"modelInstanceVersion","isSourceIdPinned":true,"modelInstanceId":385309,"modelId":404547}],"dockerImageVersionId":31041,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# **CLASSIFER with test case**","metadata":{}},{"cell_type":"code","source":"import os\nimport pandas as pd\nfrom sklearn.model_selection import StratifiedShuffleSplit, StratifiedKFold\nfrom pathlib import Path\n\ndef split_train_val_test_5fold(\n    condition: str,\n    test_size: float = 0.2,\n    n_splits: int = 2,\n    random_state: int = 42\n):\n    \"\"\"\n    1) Reads /kaggle/input/csv-files/<condition>.csv, drops NaN scores\n    2) Splits off a stratified test set (test_size fraction)\n    3) Performs a stratified n_splits-fold CV on the remaining data\n    4) Writes out:\n       - ./<condition>/test/<condition>_test.csv\n       - ./<condition>/<condition>_trainval_<n_splits>folds.csv\n       - ./<condition>/fold_0/... fold_{n_splits-1}\n    \"\"\"\n    # 1) Load and clean\n    df = pd.read_csv(f'/kaggle/input/csv-files/{condition}.csv')\n    df = df.dropna(subset=['score']).reset_index(drop=True)\n\n    # 2) Stratified shuffle split → train_val vs test\n    sss = StratifiedShuffleSplit(n_splits=1, test_size=test_size, random_state=random_state)\n    train_val_idx, test_idx = next(sss.split(df, df['score']))\n    df_train_val = df.loc[train_val_idx].reset_index(drop=True)\n    df_test      = df.loc[test_idx].reset_index(drop=True)\n\n    # Save test set\n    condition_dir = Path(f'./{condition}')\n    (condition_dir / 'test').mkdir(parents=True, exist_ok=True)\n    df_test.to_csv(\n        condition_dir / 'test' / f'{condition}_test.csv', \n        index=False\n    )\n    print(f\"Saved test set ({len(df_test)} rows) to: {condition_dir/'test'/f'{condition}_test.csv'}\")\n\n    # 3) Stratified K-Fold on train_val\n    skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=random_state)\n    df_train_val['fold'] = -1\n    for fold, (_, val_idx) in enumerate(skf.split(df_train_val, df_train_val['score'])):\n        df_train_val.loc[val_idx, 'fold'] = fold\n\n    # Save combined train_val folds CSV\n    df_train_val.to_csv(\n        condition_dir / f'{condition}_trainval_{n_splits}folds.csv',\n        index=False\n    )\n    print(f\"Saved train/val folds CSV to: {condition_dir/f'{condition}_trainval_{n_splits}folds.csv'}\")\n\n    # 4) Write out per-fold train/val splits\n    for fold in range(n_splits):\n        fold_dir = condition_dir / f'fold_{fold}'\n        fold_dir.mkdir(parents=True, exist_ok=True)\n\n        train_df = df_train_val[df_train_val['fold'] != fold].reset_index(drop=True)\n        val_df   = df_train_val[df_train_val['fold'] == fold].reset_index(drop=True)\n\n        train_df.to_csv(fold_dir / f'{condition}_train.csv', index=False)\n        val_df.to_csv(  fold_dir / f'{condition}_val.csv',   index=False)\n        print(f\"  Fold {fold}: {len(train_df)} train rows, {len(val_df)} val rows\")\n\n# Run for each condition\nfor cond in ['Neural_Foraminal_Narrowing', 'Spinal_Canal_Stenosis', 'Subarticular_Stenosis']:\n    split_train_val_test_5fold(cond)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T09:41:14.476001Z","iopub.execute_input":"2025-07-21T09:41:14.476242Z","iopub.status.idle":"2025-07-21T09:41:18.446812Z","shell.execute_reply.started":"2025-07-21T09:41:14.476217Z","shell.execute_reply":"2025-07-21T09:41:18.446055Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport pandas as pd\nimport numpy as np\nimport random\nimport shutil\n\n# Define augmentation methods\naugmentations = ['rotate', 'horizontal_flip', 'vertical_flip', 'gaussian_noise']\n\ndef augment_data(df: pd.DataFrame, augmentations: list) -> pd.DataFrame:\n    \"\"\"\n    Oversample minority \"score\" classes to balance against the majority class.\n    Adds an 'augmentation' column to each augmented sample.\n    \"\"\"\n    counts = df['score'].value_counts()\n    maj_cls = counts.idxmax()\n    target1 = counts[maj_cls] // 3\n    target2 = counts[maj_cls] // 2\n\n    minors = [cls for cls, c in counts.items() if c < counts[maj_cls]]\n    if len(minors) < 2:\n        minors = minors * 2\n\n    def sample_and_label(min_cls, target):\n        existing = df[df['score'] == min_cls]\n        needed = max(0, target - len(existing))\n        if needed <= 0:\n            return pd.DataFrame(columns=list(df.columns) + ['augmentation'])\n        sampled = existing.sample(needed, replace=True).copy()\n        sampled['augmentation'] = [random.choice(augmentations) for _ in range(needed)]\n        return sampled\n\n    aug1 = sample_and_label(minors[0], target1)\n    aug2 = sample_and_label(minors[1], target2)\n\n    df = df.copy()\n    df['augmentation'] = None\n    return pd.concat([df, aug1, aug2], ignore_index=True)\n\n# --- Adjusted for train/val/test splits ---\nconditions = ['Neural_Foraminal_Narrowing',\n              'Spinal_Canal_Stenosis',\n              'Subarticular_Stenosis']\nfolds = list(range(2))  # 5-fold CV\n\nfor cond in conditions:\n    # Augment each train fold and copy its val split\n    for fold in folds:\n        base_in  = f'./{cond}/fold_{fold}'\n        base_out = f'./augmented_output/{cond}/fold_{fold}'\n        os.makedirs(base_out, exist_ok=True)\n\n        # load train, augment, save\n        train_csv = os.path.join(base_in, f'{cond}_train.csv')\n        df_train  = pd.read_csv(train_csv)\n        df_aug    = augment_data(df_train, augmentations)\n        df_aug.to_csv(os.path.join(base_out, f'{cond}_augmented_train.csv'), index=False)\n        print(f'→ Augmented train saved: {base_out}/{cond}_augmented_train.csv')\n\n        # copy val \n        val_csv = os.path.join(base_in, f'{cond}_val.csv')\n        shutil.copy(val_csv, os.path.join(base_out, f'{cond}_val.csv'))\n        print(f'→ Val copied:           {base_out}/{cond}_val.csv')\n\n    # Copy test set\n    test_in  = f'./{cond}/test/{cond}_test.csv'\n    test_out = f'./augmented_output/{cond}/test'\n    os.makedirs(test_out, exist_ok=True)\n    shutil.copy(test_in, os.path.join(test_out, f'{cond}_test.csv'))\n    print(f'→ Test copied:          {test_out}/{cond}_test.csv')\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T09:41:21.196297Z","iopub.execute_input":"2025-07-21T09:41:21.197034Z","iopub.status.idle":"2025-07-21T09:41:21.729677Z","shell.execute_reply.started":"2025-07-21T09:41:21.197001Z","shell.execute_reply":"2025-07-21T09:41:21.728937Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ─────────────────────────────────────────────────────────────────────────────\n# Cell: Image Data Preparation for 5-Fold CV + Held-Out Test\n# ─────────────────────────────────────────────────────────────────────────────\n\nimport os\nimport pydicom\nimport pandas as pd\nimport numpy as np\nfrom PIL import Image, ImageOps\nfrom pathlib import Path\n\nclass DataPreparationImage:\n    def __init__(\n        self,\n        dataset_directory: str,\n        condition: str,\n        csv_directory: str,\n        num_folds: int = 2,\n        augmentation_list=None,\n    ):\n        self.dataset_directory = Path(dataset_directory)\n        self.condition       = condition\n        self.csv_directory   = Path(csv_directory)\n        self.num_folds       = num_folds\n        self.augmentation_list = augmentation_list or [\n            'rotate', 'horizontal_flip', 'vertical_flip', 'gaussian_noise'\n        ]\n\n        print(f\"Starting image data prep for {self.condition} ({self.num_folds} folds + test)\")\n        self._create_folders()\n        self._process_all_folds()\n        self._process_test()\n\n    def _create_folders(self):\n        base = Path(f\"./{self.condition}\")\n        base.mkdir(exist_ok=True)\n        # train / val\n        for fold in range(self.num_folds):\n            (base / f\"fold_{fold}\" / \"train\").mkdir(parents=True, exist_ok=True)\n            (base / f\"fold_{fold}\" / \"val\").mkdir(parents=True, exist_ok=True)\n        # held-out test\n        (base / \"test\").mkdir(parents=True, exist_ok=True)\n\n    def _read_csv(self, split: str, fold: int = None) -> pd.DataFrame:\n        \"\"\"\n        split: 'train', 'val', or 'test'\n        fold: required for train/val, ignored for test\n        \"\"\"\n        if split in ('train', 'val'):\n            if fold is None:\n                raise ValueError(\"Must provide fold for train/val\")\n            suffix = 'augmented_train' if split == 'train' else 'val'\n            filename = f\"{self.condition}_{suffix}.csv\"\n            path = self.csv_directory / self.condition / f\"fold_{fold}\" / filename\n        elif split == 'test':\n            filename = f\"{self.condition}_test.csv\"\n            path = self.csv_directory / self.condition / \"test\" / filename\n        else:\n            raise ValueError(f\"Unknown split: {split}\")\n        return pd.read_csv(path)\n\n    def _read_dicom(self, path: Path) -> np.ndarray:\n        ds = pydicom.dcmread(str(path))\n        img = ds.pixel_array.astype(float)\n        img = (img - img.min()) / (img.max() - img.min() + 1e-6) * 255.0\n        rgb = np.stack([img] * 3, axis=-1)\n        return rgb.astype('uint8')\n\n    def _crop(self, image: np.ndarray, x: float, y: float, box: int = 64) -> Image.Image:\n        \"\"\"\n        Crop a (2*box)x(2*box)=128x128 patch centered at (x,y),\n        then resize to 224x224 with bicubic interpolation.\n        \"\"\"\n        img = Image.fromarray(image)\n        left, top   = int(x - box), int(y - box)\n        right, bottom = int(x + box), int(y + box)\n        patch = img.crop((left, top, right, bottom))\n        return patch.resize((224, 224), Image.BICUBIC)\n    \n\n    def _apply_augmentation(self, img: Image.Image, aug: str) -> Image.Image:\n        if aug == 'rotate':\n            return img.rotate(np.random.uniform(-20,20), expand=True)\n        if aug == 'horizontal_flip':\n            return ImageOps.mirror(img)\n        if aug == 'vertical_flip':\n            return ImageOps.flip(img)\n        if aug == 'gaussian_noise':\n            arr   = np.array(img)\n            noise = np.random.normal(0,25,arr.shape)\n            return Image.fromarray(np.clip(arr+noise,0,255).astype('uint8'))\n        return img\n\n    def _process_all_folds(self):\n        for fold in range(self.num_folds):\n            print(f\" Processing fold {fold}…\")\n            # TRAIN (with augmentation)\n            df_train = self._read_csv('train', fold)\n            for _, row in df_train.iterrows():\n                sid, seid, inst = row['study_id'], row['series_id'], row['instance_number']\n                x, y    = row['x'], row['y']\n                aug_op  = row.get('augmentation')\n                dcm     = self.dataset_directory / str(sid) / str(seid) / f\"{inst}.dcm\"\n                img     = self._read_dicom(dcm)\n                patch   = self._crop(img, x, y)\n                if aug_op in self.augmentation_list:\n                    out_img = self._apply_augmentation(patch, aug_op)\n                    suffix  = f\"_{aug_op}\"\n                else:\n                    out_img = patch\n                    suffix  = \"\"\n                fname = f\"{sid}_{seid}_{inst}_{int(x)}_{int(y)}{suffix}.png\"\n                out_path = Path(self.condition) / f\"fold_{fold}\" / \"train\" / fname\n                out_img.save(out_path)\n\n            # VAL (no augmentation)\n            df_val = self._read_csv('val', fold)\n            for _, row in df_val.iterrows():\n                sid, seid, inst = row['study_id'], row['series_id'], row['instance_number']\n                x, y    = row['x'], row['y']\n                dcm     = self.dataset_directory / str(sid) / str(seid) / f\"{inst}.dcm\"\n                img     = self._read_dicom(dcm)\n                patch   = self._crop(img, x, y)\n                fname   = f\"{sid}_{seid}_{inst}_{int(x)}_{int(y)}.png\"\n                out_path = Path(self.condition) / f\"fold_{fold}\" / \"val\" / fname\n                patch.save(out_path)\n\n    def _process_test(self):\n        print(\" Processing held-out test set…\")\n        df_test = self._read_csv('test')\n        for _, row in df_test.iterrows():\n            sid, seid, inst = row['study_id'], row['series_id'], row['instance_number']\n            x, y    = row['x'], row['y']\n            dcm     = self.dataset_directory / str(sid) / str(seid) / f\"{inst}.dcm\"\n            img     = self._read_dicom(dcm)\n            patch   = self._crop(img, x, y)\n            fname   = f\"{sid}_{seid}_{inst}_{int(x)}_{int(y)}.png\"\n            out_path = Path(self.condition) / \"test\" / fname\n            patch.save(out_path)\n        print(f\" Saved {len(df_test)} test images to ./{self.condition}/test\")\n\n\n# ─────────────────────────────────────────────────────────────────────────────\n# Run for your condition (example: Neural Foraminal Narrowing)\n# ─────────────────────────────────────────────────────────────────────────────\n\nDATASET_DIR = \"/kaggle/input/rsna-2024-lumbar-spine-degenerative-classification/train_images\"\nCSV_DIR     = \"/kaggle/working/augmented_output\"\n\nDataPreparationImage(\n    dataset_directory=DATASET_DIR,\n    condition=\"Neural_Foraminal_Narrowing\",\n    csv_directory=CSV_DIR,\n    num_folds=2\n)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T09:42:45.999173Z","iopub.execute_input":"2025-07-21T09:42:45.999453Z","iopub.status.idle":"2025-07-21T10:00:18.189957Z","shell.execute_reply.started":"2025-07-21T09:42:45.999432Z","shell.execute_reply":"2025-07-21T10:00:18.189368Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport pandas as pd\nfrom pathlib import Path\n\n# 1) Parameters\nCSV_DIR    = Path('/kaggle/working/augmented_output')   # where your augmented CSVs live\nCONDITIONS = [\n    'Spinal_Canal_Stenosis',\n    'Neural_Foraminal_Narrowing',\n    'Subarticular_Stenosis'\n]\nFOLDS      = list(range(2))  # five folds: 0–4\nOUT_ROOT   = Path('/kaggle/working')    # base for label output folders\n\n# 2) Helper: build 'subject' and 'label' columns\ndef working_on_csv(csv_path: Path, split: str) -> pd.DataFrame:\n    df = pd.read_csv(csv_path)\n\n    # ensure we always have an 'augmentation' column\n    if 'augmentation' not in df.columns:\n        df['augmentation'] = None\n\n    def make_subject(r):\n        name = f\"{r.study_id}_{r.series_id}_{r.instance_number}_{int(r.x)}_{int(r.y)}.png\"\n        # only append \"_augmented\" on training samples that were actually augmented\n        if split == 'train' and pd.notna(r.augmentation) and r.augmentation:\n            name = name.replace('.png', '_augmented.png')\n        return name\n\n    def make_label(score):\n        # if your original CSV already used 1,2,3 as numeric scores, just cast\n        if pd.api.types.is_numeric_dtype(type(score)):\n            return int(score)\n        # otherwise map from the text labels you used\n        return {\n            'Normal/Mild': 1,\n            'Moderate':    2,\n            'Severe':      3\n        }.get(score, None)\n\n    df['subject'] = df.apply(make_subject, axis=1)\n    df['label']   = df['score'].apply(make_label)\n\n    return df[['subject', 'label']]\n\n# 3) Create output folder structure (including test)\nfor cond in CONDITIONS:\n    base = OUT_ROOT / f\"{cond}_label\"\n    for fold in FOLDS:\n        (base / f\"fold_{fold}\").mkdir(parents=True, exist_ok=True)\n    (base / \"test\").mkdir(parents=True, exist_ok=True)\n\n# 4) Process each condition: train/val and then test\nfor cond in CONDITIONS:\n    label_root = OUT_ROOT / f\"{cond}_label\"\n\n    # train & val\n    for fold in FOLDS:\n        train_csv = CSV_DIR / cond / f\"fold_{fold}\" / f\"{cond}_augmented_train.csv\"\n        val_csv   = CSV_DIR / cond / f\"fold_{fold}\" / f\"{cond}_val.csv\"\n\n        train_labels = working_on_csv(train_csv, 'train')\n        val_labels   = working_on_csv(val_csv,   'val')\n\n        out_base = label_root / f\"fold_{fold}\"\n        train_labels.to_csv(out_base / f\"{cond}_augmented_labels.csv\", index=False)\n        val_labels.to_csv(  out_base / f\"{cond}_val_labels.csv\",       index=False)\n\n        print(f\"→ {cond} fold {fold}: train={len(train_labels)}, val={len(val_labels)}\")\n\n    # test\n    test_csv    = CSV_DIR / cond / \"test\" / f\"{cond}_test.csv\"\n    test_labels = working_on_csv(test_csv, 'test')\n    out_test    = label_root / \"test\"\n    test_labels.to_csv(out_test / f\"{cond}_test_labels.csv\", index=False)\n\n    print(f\"→ {cond} test: {len(test_labels)} rows → {out_test}/{cond}_test_labels.csv\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T10:00:18.191186Z","iopub.execute_input":"2025-07-21T10:00:18.191648Z","iopub.status.idle":"2025-07-21T10:00:21.425517Z","shell.execute_reply.started":"2025-07-21T10:00:18.191630Z","shell.execute_reply":"2025-07-21T10:00:21.424869Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Stage 1 - Normal and Abnormal Cases ","metadata":{}},{"cell_type":"code","source":"# ─────────────────────────────────────────────────────────────────────────────\n# Cell: 5-Fold CV + Held-Out Test Evaluation for Neural Foraminal Narrowing\n#           (50 layers frozen, Flatten-only head, L2 on hidden layers, no reg on final)\n# ─────────────────────────────────────────────────────────────────────────────\n\nimport os\nimport gc\nimport json\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom tensorflow.keras.applications import DenseNet121\nimport tensorflow as tf\nimport tensorflow.keras.backend as K\nfrom tensorflow.keras.applications import ResNet50\nfrom tensorflow.keras.models import Model, load_model\nfrom tensorflow.keras.layers import Flatten, Dense, Dropout, BatchNormalization\nfrom tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\nfrom tensorflow.keras.regularizers import l2\n\nfrom sklearn.metrics import (\n    accuracy_score, precision_score, recall_score, f1_score,\n    confusion_matrix, roc_curve, auc\n)\nfrom sklearn.preprocessing import label_binarize\nimport matplotlib.pyplot as plt\n\n# ── Configuration ────────────────────────────────────────────────────────────\nCONDITION       = 'Neural_Foraminal_Narrowing'\nFOLDS           = list(range(2))\nIMAGE_ROOT_TMPL = '/kaggle/working/Neural_Foraminal_Narrowing/fold_{fold}'\nLABEL_ROOT_TMPL = '/kaggle/working/Neural_Foraminal_Narrowing_label/fold_{fold}'\nTEST_IMAGE_DIR  = f'/kaggle/working/{CONDITION}/test'\nTEST_LABEL_CSV  = f'/kaggle/working/{CONDITION}_label/test/{CONDITION}_test_labels.csv'\nRESULTS_DIR     = Path(f'./results/{CONDITION}')\n\nBATCH_SIZE      = 16\nIMG_SIZE        = (224, 224)\nEPOCHS          = 10\nPATIENCE        = 5\nLEARNING_RATE   = 1e-4\nL2_WEIGHT       = 1e-4\nDROP_RATE       = 0.5\n\n# ── Focal Loss (per-class α) ──────────────────────────────────────────────────\ndef focal_loss(gamma=2.0, alpha=None):\n    \"\"\"\n    gamma: focusing parameter\n    alpha: None or list of length num_classes\n    \"\"\"\n    def loss_fn(y_true, y_pred):\n        eps    = K.epsilon()\n        y_pred = K.clip(y_pred, eps, 1. - eps)\n        ce     = -y_true * K.log(y_pred)              # cross-entropy\n        mod    = K.pow(1. - y_pred, gamma)            # modulating factor\n\n        if alpha is not None:\n            alpha_tensor = tf.constant(alpha, dtype=tf.float32)\n            # only true class gets its alpha weight\n            alpha_factor = y_true * alpha_tensor\n            ce = ce * alpha_factor\n\n        loss = mod * ce\n        return K.sum(loss, axis=-1)                   # sum over classes\n    return loss_fn\n\nfold_results = []\n\nfor fold in FOLDS:\n    print(f\"\\n▶ Training fold {fold}\")\n    IMAGE_ROOT = IMAGE_ROOT_TMPL.format(fold=fold)\n    LABEL_ROOT = LABEL_ROOT_TMPL.format(fold=fold)\n\n    # — Load & preprocess labels\n    train_df = pd.read_csv(Path(LABEL_ROOT) / f'{CONDITION}_augmented_labels.csv')\n    val_df   = pd.read_csv(Path(LABEL_ROOT) / f'{CONDITION}_val_labels.csv')\n    for df in (train_df, val_df):\n        df['subject'] = df['subject'].str.replace('_augmented','',regex=False)\n        df['label']   = (df['label'].astype(int) - 1).astype(str)\n        df['label_bin'] = (df['label'].apply(lambda x: 0 if int(x)==0 else 1)).astype(str)\n        \n\n    y_train_bin = train_df['label_bin'].astype(str)\n    counts      = np.bincount(y_train_bin, minlength=2)\n    total       = len(y_train_bin)\n    n_classes   = len(counts)\n    class_weight = {i: total/(n_classes * counts[i]) for i in range(n_classes)}\n\n\n    # — Compute per-class α for focal loss\n    freqs = counts / total\n    alpha = ((1 - freqs) / (1 - freqs).sum()).tolist()\n\n    # — Data generators (only rescale)\n    datagen    = ImageDataGenerator(rescale=1./255)\n    train_gen  = datagen.flow_from_dataframe(\n        train_df, directory=os.path.join(IMAGE_ROOT,'train'),\n        x_col='subject', y_col='label_bin',\n        target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n        class_mode='categorical', shuffle=True\n    )\n    val_gen    = datagen.flow_from_dataframe(\n        val_df, directory=os.path.join(IMAGE_ROOT,'val'),\n        x_col='subject', y_col='label_bin',\n        target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n        class_mode='categorical', shuffle=False\n    )\n    \n    base = DenseNet121(\n        include_top=False,\n        weights=None,\n        input_shape=IMG_SIZE + (3,)\n    )\n    base.load_weights('/kaggle/input/rad/keras/default/1/RadImageNet-DenseNet121_notop.h5')\n\n\n    x = Flatten()(base.output)\n    x = Dense(256, activation='relu', kernel_regularizer=l2(L2_WEIGHT))(x)\n    x = BatchNormalization()(x)\n    x = Dropout(DROP_RATE)(x)\n    x = Dense(128, activation='relu', kernel_regularizer=l2(L2_WEIGHT))(x)\n    x = Dropout(DROP_RATE)(x)\n    preds = Dense(2, activation='softmax')(x)\n\n    model_stage1 = Model(inputs=base.input, outputs=preds)\n    model_stage1.compile(\n        optimizer=tf.keras.optimizers.Adam(learning_rate=LEARNING_RATE),\n        loss=focal_loss(gamma=2.0, alpha=alpha),\n        metrics=['accuracy']\n    )\n\n    # — Callbacks\n    out_dir = RESULTS_DIR / f'fold_{fold}'\n    out_dir.mkdir(parents=True, exist_ok=True)\n    ckpt = ModelCheckpoint(out_dir/'best_model.keras', monitor='val_loss',\n                           save_best_only=True, verbose=1)\n    es   = EarlyStopping(monitor='val_loss', patience=PATIENCE,\n                         restore_best_weights=True, verbose=1)\n    rl   = ReduceLROnPlateau(monitor='val_loss', factor=0.5,\n                             patience=2, min_lr=1e-6, verbose=1)\n\n    # — Train with class_weight\n    history = model_stage1.fit(\n        train_gen,\n        validation_data=val_gen,\n        epochs=EPOCHS,\n        callbacks=[es, ckpt, rl],\n        class_weight=class_weight,\n        verbose=2\n    )\n    # — Save history\n    with open(out_dir/'history.json','w') as f:\n        json.dump(history.history, f)\n\n    # — Evaluate on validation\n    val_gen.reset()\n    y_pred = np.argmax(model_stage1.predict(val_gen), axis=1)\n    y_true = val_gen.classes\n    fm = {\n        'fold': fold,\n        'Accuracy':  accuracy_score(y_true, y_pred),\n        'Precision': precision_score(y_true, y_pred, average='weighted'),\n        'Recall':    recall_score(y_true, y_pred, average='weighted'),\n        'F1 Score':  f1_score(y_true, y_pred, average='weighted')\n    }\n    fold_results.append(fm)\n    with open(out_dir/'metrics.txt','w') as f:\n        for k,v in fm.items():\n            if k!='fold': f.write(f\"{k}: {v:.4f}\\n\")\n    print(f\"✅ Fold {fold} metrics:\", fm)\n\n    # — Cleanup\n    tf.keras.backend.clear_session()\n    gc.collect()\n\n\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T10:30:21.961568Z","iopub.execute_input":"2025-07-21T10:30:21.962239Z","iopub.status.idle":"2025-07-21T11:10:49.467480Z","shell.execute_reply.started":"2025-07-21T10:30:21.962216Z","shell.execute_reply":"2025-07-21T11:10:49.466600Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Stage 2 - Severe and Moderate Cases","metadata":{}},{"cell_type":"code","source":"# ─────────────────────────────────────────────────────────────────────────────\n# Cell: 5-Fold CV + Held-Out Test Evaluation for Neural Foraminal Narrowing\n#           (50 layers frozen, Flatten-only head, L2 on hidden layers, no reg on final)\n# ─────────────────────────────────────────────────────────────────────────────\n\nimport os\nimport gc\nimport json\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nfrom tensorflow.keras.applications import DenseNet121\nimport tensorflow as tf\nimport tensorflow.keras.backend as K\nfrom tensorflow.keras.applications import ResNet50\nfrom tensorflow.keras.models import Model, load_model\nfrom tensorflow.keras.layers import Flatten, Dense, Dropout, BatchNormalization\nfrom tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\nfrom tensorflow.keras.regularizers import l2\n\nfrom sklearn.metrics import (\n    accuracy_score, precision_score, recall_score, f1_score,\n    confusion_matrix, roc_curve, auc\n)\nfrom sklearn.preprocessing import label_binarize\nimport matplotlib.pyplot as plt\n\n# ── Configuration ────────────────────────────────────────────────────────────\nCONDITION       = 'Neural_Foraminal_Narrowing'\nFOLDS           = list(range(2))\nIMAGE_ROOT_TMPL = '/kaggle/working/Neural_Foraminal_Narrowing/fold_{fold}'\nLABEL_ROOT_TMPL = '/kaggle/working/Neural_Foraminal_Narrowing_label/fold_{fold}'\nTEST_IMAGE_DIR  = f'/kaggle/working/{CONDITION}/test'\nTEST_LABEL_CSV  = f'/kaggle/working/{CONDITION}_label/test/{CONDITION}_test_labels.csv'\nRESULTS_DIR     = Path(f'./results/{CONDITION}')\n\nBATCH_SIZE      = 16\nIMG_SIZE        = (224, 224)\nEPOCHS          = 10\nPATIENCE        = 5\nLEARNING_RATE   = 1e-4\nL2_WEIGHT       = 1e-4\nDROP_RATE       = 0.5\n\n# ── Focal Loss (per-class α) ──────────────────────────────────────────────────\ndef focal_loss(gamma=2.0, alpha=None):\n    \"\"\"\n    gamma: focusing parameter\n    alpha: None or list of length num_classes\n    \"\"\"\n    def loss_fn(y_true, y_pred):\n        eps    = K.epsilon()\n        y_pred = K.clip(y_pred, eps, 1. - eps)\n        ce     = -y_true * K.log(y_pred)              # cross-entropy\n        mod    = K.pow(1. - y_pred, gamma)            # modulating factor\n\n        if alpha is not None:\n            alpha_tensor = tf.constant(alpha, dtype=tf.float32)\n            # only true class gets its alpha weight\n            alpha_factor = y_true * alpha_tensor\n            ce = ce * alpha_factor\n\n        loss = mod * ce\n        return K.sum(loss, axis=-1)                   # sum over classes\n    return loss_fn\n\nfold_results = []\n\nfor fold in FOLDS:\n    print(f\"\\n▶ Training fold {fold}\")\n    IMAGE_ROOT = IMAGE_ROOT_TMPL.format(fold=fold)\n    LABEL_ROOT = LABEL_ROOT_TMPL.format(fold=fold)\n\n    # 1. Load fresh each fold to avoid remapping on already-remapped columns\n    train_df = pd.read_csv(Path(LABEL_ROOT) / f'{CONDITION}_augmented_labels.csv')\n    val_df   = pd.read_csv(Path(LABEL_ROOT) / f'{CONDITION}_val_labels.csv')\n\n    # 2. Initial mapping: subtract 1, only ONCE (so [1,2] -> [0,1], normal is 0)\n    for df in (train_df, val_df):\n        df['subject'] = df['subject'].str.replace('_augmented','',regex=False)\n        df['label'] = df['label'].astype(int) - 1   # Now: 0 (normal), 1, 2\n        df['label'] = df['label'].astype(str)       \n        df['label_bin'] = df['label'].apply(lambda x: 0 if x == \"0\" else 1).astype(str)\n\n    # 3. Filter abnormal\n    train_abn_df = train_df[train_df['label_bin'] == \"1\"].copy()\n    val_abn_df   = val_df[val_df['label_bin'] == \"1\"].copy()\n\n    # 4. Remap abnormals: [1,2] (as strings) to [0,1] (as strings) for stage 2!\n    train_abn_df['label'] = train_abn_df['label'].astype(int) - 1\n    val_abn_df['label']   = val_abn_df['label'].astype(int) - 1\n    train_abn_df['label'] = train_abn_df['label'].astype(str)\n    val_abn_df['label']   = val_abn_df['label'].astype(str)\n\n    print(\"Unique labels for abnormal train after remap:\", train_abn_df['label'].unique())\n\n    # 5. No negatives! Now y_train_abn is only 0/1\n    y_train_abn = train_abn_df['label'].astype(int).values\n    counts = np.bincount(y_train_abn, minlength=2)\n    total = len(y_train_abn)\n    n_classes = len(counts)\n    print(f\"Counts for fold {fold}: {counts}\")\n\n    if np.any(counts == 0):\n        print(f\"WARNING: Missing one or more abnormal classes in fold {fold}: counts={counts}\")\n        class_weight = None\n        alpha = [0.5, 0.5]\n    else:\n        class_weight = {i: total/(n_classes * counts[i]) for i in range(n_classes)}\n        freqs = counts / total\n        alpha = ((1 - freqs) / (1 - freqs).sum()).tolist()\n    print(f\"Stage 2 alpha: {alpha}\")\n\n    # — Data generators (only rescale)\n    datagen    = ImageDataGenerator(rescale=1./255)\n    train_gen  = datagen.flow_from_dataframe(\n        train_abn_df, directory=os.path.join(IMAGE_ROOT,'train'),\n        x_col='subject', y_col='label',\n        target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n        class_mode='categorical', shuffle=True\n    )\n    val_gen    = datagen.flow_from_dataframe(\n        val_abn_df, directory=os.path.join(IMAGE_ROOT,'val'),\n        x_col='subject', y_col='label',\n        target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n        class_mode='categorical', shuffle=False\n    )\n\n\n\n    base = DenseNet121(\n        include_top=False,\n        weights=None,\n        input_shape=IMG_SIZE + (3,)\n    )\n    base.load_weights('/kaggle/input/rad/keras/default/1/RadImageNet-DenseNet121_notop.h5')\n\n\n    x = Flatten()(base.output)\n    x = Dense(256, activation='relu', kernel_regularizer=l2(L2_WEIGHT))(x)\n    x = BatchNormalization()(x)\n    x = Dropout(DROP_RATE)(x)\n    x = Dense(128, activation='relu', kernel_regularizer=l2(L2_WEIGHT))(x)\n    x = Dropout(DROP_RATE)(x)\n    preds = Dense(2, activation='softmax')(x)\n\n    model_stage2 = Model(inputs=base.input, outputs=preds)\n    model_stage2.compile(\n        optimizer=tf.keras.optimizers.Adam(learning_rate=LEARNING_RATE),\n        loss=focal_loss(gamma=2.0, alpha=alpha),\n        metrics=['accuracy']\n    )\n\n    # — Callbacks\n    out_dir = RESULTS_DIR / f'fold_{fold}_stage2'\n    out_dir.mkdir(parents=True, exist_ok=True)\n    ckpt = ModelCheckpoint(out_dir/'best_model.keras', monitor='val_loss',\n                           save_best_only=True, verbose=1)\n    es   = EarlyStopping(monitor='val_loss', patience=PATIENCE,\n                         restore_best_weights=True, verbose=1)\n    rl   = ReduceLROnPlateau(monitor='val_loss', factor=0.5,\n                             patience=2, min_lr=1e-6, verbose=1)\n\n    # — Train with class_weight\n    history = model_stage2.fit(\n        train_gen,\n        validation_data=val_gen,\n        epochs=EPOCHS,\n        callbacks=[es, ckpt, rl],\n        class_weight=class_weight,\n        verbose=2\n    )\n    # — Save history\n    with open(out_dir/'history.json','w') as f:\n        json.dump(history.history, f)\n\n    # — Evaluate on validation\n    val_gen.reset()\n    y_pred = np.argmax(model_stage2.predict(val_gen), axis=1)\n    y_true = val_gen.classes\n    fm = {\n        'fold': fold,\n        'Accuracy':  accuracy_score(y_true, y_pred),\n        'Precision': precision_score(y_true, y_pred, average='weighted'),\n        'Recall':    recall_score(y_true, y_pred, average='weighted'),\n        'F1 Score':  f1_score(y_true, y_pred, average='weighted')\n    }\n    fold_results.append(fm)\n    with open(out_dir/'metrics.txt','w') as f:\n        for k,v in fm.items():\n            if k!='fold': f.write(f\"{k}: {v:.4f}\\n\")\n    print(f\"✅ Fold {fold} metrics:\", fm)\n\n    # — Cleanup\n    tf.keras.backend.clear_session()\n    gc.collect()\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T12:03:39.439529Z","iopub.execute_input":"2025-07-21T12:03:39.440011Z","iopub.status.idle":"2025-07-21T12:21:48.144098Z","shell.execute_reply.started":"2025-07-21T12:03:39.439989Z","shell.execute_reply":"2025-07-21T12:21:48.143473Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Testing","metadata":{}},{"cell_type":"code","source":"\nfrom tensorflow.keras.utils import img_to_array, load_img\nbest_fold = 1\n# --- Load best Stage 1 and Stage 2 models ---\nmodel_stage1 = load_model(\n    RESULTS_DIR/f'fold_{best_fold}'/'best_model.keras',\n    custom_objects={'loss_fn': focal_loss(gamma=2.0, alpha=alpha)}\n)\nmodel_stage2 = load_model(\n    RESULTS_DIR/f'fold_{best_fold}_stage2'/'best_model.keras',\n    custom_objects={'loss_fn': focal_loss(gamma=2.0, alpha=alpha)}\n)\n\n# --- Load test DataFrame ---\ntest_df = pd.read_csv(TEST_LABEL_CSV)\ntest_df['label'] = (test_df['label'].astype(int) - 1).astype(str)\ntest_df['label_bin'] = test_df['label'].apply(lambda x: 0 if int(x)=='0' else 1)\n\n# --- Prepare image paths ---\ndef get_img_path(row):\n    return os.path.join(TEST_IMAGE_DIR, row['subject'])\ntest_df['img_path'] = test_df.apply(get_img_path, axis=1)\n\nn_normal = 800\nnormal_df    = test_df[test_df['label'] == \"0\"].sample(n=n_normal, random_state=42)\nmoderate_df  = test_df[test_df['label'] == \"1\"]\nsevere_df    = test_df[test_df['label'] == \"2\"]\n\n# Concatenate to get new test set\ntest_df_sub = pd.concat([normal_df, moderate_df, severe_df], ignore_index=True)\n# (optional) shuffle for randomness\ntest_df_sub = test_df_sub.sample(frac=1, random_state=42).reset_index(drop=True)\n\n# Then use test_df_sub in your inference loop:\ntest_df = test_df_sub\n\n# --- Inference loop ---\ny_true = []\ny_pred = []\n\nfor _, row in test_df.iterrows():\n    # Load and preprocess image\n    img = load_img(row['img_path'], target_size=IMG_SIZE)\n    x = img_to_array(img) / 255.0\n    x = np.expand_dims(x, axis=0)\n\n    # Stage 1 prediction (normal/abnormal)\n    pred1 = np.argmax(model_stage1.predict(x, verbose=0), axis=1)[0]\n    if pred1 == 0:\n        y_pred.append(0)  # Normal\n    else:\n        # Stage 2 prediction (among abnormals)\n        pred2 = np.argmax(model_stage2.predict(x, verbose=0), axis=1)[0]\n        # Stage 2 outputs 0 or 1; map back to original abnormal labels (1 or 2)\n        y_pred.append(pred2 + 1)\n    y_true.append(int(row['label']))\n\n# --- Evaluate ---\nacc = accuracy_score(y_true, y_pred)\nprec = precision_score(y_true, y_pred, average='weighted')\nrec = recall_score(y_true, y_pred, average='weighted')\nf1 = f1_score(y_true, y_pred, average='weighted')\ncm = confusion_matrix(y_true, y_pred)\n\nprint(f\"Test Accuracy: {acc:.4f}\")\nprint(f\"Test Precision: {prec:.4f}\")\nprint(f\"Test Recall: {rec:.4f}\")\nprint(f\"Test F1: {f1:.4f}\")\nprint(\"Confusion Matrix:\")\nprint(cm)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T12:41:05.052120Z","iopub.execute_input":"2025-07-21T12:41:05.052389Z","iopub.status.idle":"2025-07-21T12:44:58.824683Z","shell.execute_reply.started":"2025-07-21T12:41:05.052373Z","shell.execute_reply":"2025-07-21T12:44:58.823894Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"from sklearn.metrics import ConfusionMatrixDisplay\nfrom sklearn.metrics import classification_report\n\nmetrics_test = {\n    'Accuracy':  accuracy_score(y_true, y_pred),\n    'Precision': precision_score(y_true, y_pred, average='weighted'),\n    'Recall':    recall_score(y_true, y_pred, average='weighted'),\n    'F1 Score':  f1_score(y_true, y_pred, average='weighted'),\n    'Error':     1 - accuracy_score(y_true, y_pred)\n}\nprint(\"\\n➡️ Held-out TEST metrics:\")\nfor k,v in metrics_test.items():\n    print(f\"   {k}: {v:.3f}\")\n\n# Confusion matrix\ncm = confusion_matrix(y_true, y_pred)\nlabels = ['Normal', 'Moderate', 'Severe']  # adjust to your actual class names if needed\ndisp = ConfusionMatrixDisplay(cm, display_labels=labels)\nplt.figure(figsize=(4,4))\ndisp.plot(ax=plt.gca(), cmap='Blues', colorbar=False)\nplt.title('Held-out Test Confusion')\nplt.show()\n\n# ROC\ny_bin = label_binarize(y_true, classes=[0,1,2])\ny_pred_prob = np.zeros((len(y_pred), 3))\n\nprint(\"\\nClassification Report:\")\nprint(classification_report(y_true, y_pred, target_names=labels, digits=3))","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-21T12:48:31.978466Z","iopub.execute_input":"2025-07-21T12:48:31.979023Z","iopub.status.idle":"2025-07-21T12:48:32.108412Z","shell.execute_reply.started":"2025-07-21T12:48:31.978999Z","shell.execute_reply":"2025-07-21T12:48:32.107803Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Evaluation Need to be done according to the 2 stage","metadata":{}},{"cell_type":"code","source":"# ─────────────────────────────────────────────────────────────────────────────\n# Cell: Summary for CV & Held-Out Test – Loss Curves, Avg Metrics, Confusion & ROC\n# ─────────────────────────────────────────────────────────────────────────────\n\nimport json, os\nimport numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport tensorflow as tf\nimport tensorflow.keras.backend as K\n\nfrom pathlib import Path\nfrom sklearn.metrics import (\n    accuracy_score, precision_score, recall_score, f1_score,\n    confusion_matrix, ConfusionMatrixDisplay, roc_curve, auc\n)\nfrom sklearn.preprocessing import label_binarize\nfrom tensorflow.keras.preprocessing.image import ImageDataGenerator\nfrom tensorflow.keras.models import load_model\n\n# ── Config ───────────────────────────────────────────────────────────────────\nCONDITION      = 'Neural_Foraminal_Narrowing'\nRESULTS_DIR    = Path(f'./results/{CONDITION}')\nLABEL_DIR      = Path(f'/kaggle/working/{CONDITION}_label')\nIMG_DIR        = Path(f'/kaggle/working/{CONDITION}')\nTEST_IMAGE_DIR = IMG_DIR / 'test'\nTEST_LABEL_CSV = LABEL_DIR / 'test' / f'{CONDITION}_test_labels.csv'\n\nBATCH_SIZE = 16\nIMG_SIZE   = (224, 224)\n\n# ── Focal Loss (per‐class α) – must exactly match training cell! ──────────────\ndef focal_loss(gamma=2.0, alpha=None):\n    \"\"\"\n    gamma: focusing parameter\n    alpha: None or list of length num_classes\n    \"\"\"\n    def loss_fn(y_true, y_pred):\n        eps    = K.epsilon()\n        y_pred = K.clip(y_pred, eps, 1. - eps)\n        ce     = -y_true * K.log(y_pred)               # cross‐entropy\n        mod    = K.pow(1. - y_pred, gamma)             # modulating factor\n\n        if alpha is not None:\n            # per‐class alpha weighting\n            alpha_tensor = tf.constant(alpha, dtype=tf.float32)\n            alpha_factor = y_true * alpha_tensor\n            ce = ce * alpha_factor\n\n        loss = mod * ce\n        return K.sum(loss, axis=-1)                    # sum over classes\n    return loss_fn\n\n# ── Gather fold dirs ─────────────────────────────────────────────────────────\nfold_dirs = sorted(RESULTS_DIR.glob('fold_*'),\n                   key=lambda p: int(p.name.split('_')[1]))\n\n# ── Containers ───────────────────────────────────────────────────────────────\ntrain_metrics, val_metrics = [], []\nloss_curves = {}\n\n# ── 1) Collect CV train/val losses & metrics ─────────────────────────────────\nfor fd in fold_dirs:\n    fold = int(fd.name.split('_')[1])\n    # a) load history\n    hist = json.load(open(fd/'history.json'))\n    loss_curves[fold] = (hist['loss'], hist['val_loss'])\n\n    # b) load model (only for naming the custom loss – won't re‐train)\n    model = load_model(\n        fd/'best_model.keras',\n        custom_objects={'loss_fn': focal_loss(gamma=2.0, alpha=None)}\n    )\n\n    # c) prepare train generator\n    train_csv = LABEL_DIR/f'fold_{fold}'/f'{CONDITION}_augmented_labels.csv'\n    df_train  = pd.read_csv(train_csv)\n    df_train['subject'] = df_train['subject'].str.replace('_augmented','',regex=False)\n    df_train['label']   = (df_train['label'].astype(int)-1).astype(str)\n    train_gen = ImageDataGenerator(rescale=1./255).flow_from_dataframe(\n        df_train,\n        directory=IMG_DIR/f'fold_{fold}'/'train',\n        x_col='subject', y_col='label',\n        target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n        class_mode='categorical', shuffle=False\n    )\n\n    # d) prepare val generator\n    val_csv = LABEL_DIR/f'fold_{fold}'/f'{CONDITION}_val_labels.csv'\n    df_val  = pd.read_csv(val_csv)\n    df_val['subject'] = df_val['subject'].str.replace('_augmented','',regex=False)\n    df_val['label']   = (df_val['label'].astype(int)-1).astype(str)\n    val_gen = ImageDataGenerator(rescale=1./255).flow_from_dataframe(\n        df_val,\n        directory=IMG_DIR/f'fold_{fold}'/'val',\n        x_col='subject', y_col='label',\n        target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n        class_mode='categorical', shuffle=False\n    )\n\n    # e) get predictions to compute metrics\n    y_tr_pred = np.argmax(model.predict(train_gen), axis=1)\n    y_tr_true = train_gen.classes\n    y_vl_pred = np.argmax(model.predict(val_gen),   axis=1)\n    y_vl_true = val_gen.classes\n\n    train_metrics.append({\n        'Accuracy':  accuracy_score(y_tr_true, y_tr_pred),\n        'Precision': precision_score(y_tr_true, y_tr_pred, average='weighted'),\n        'Recall':    recall_score(y_tr_true, y_tr_pred, average='weighted'),\n        'F1 Score':  f1_score(y_tr_true, y_tr_pred, average='weighted')\n    })\n    val_metrics.append({\n        'Accuracy':  accuracy_score(y_vl_true, y_vl_pred),\n        'Precision': precision_score(y_vl_true, y_vl_pred, average='weighted'),\n        'Recall':    recall_score(y_vl_true, y_vl_pred, average='weighted'),\n        'F1 Score':  f1_score(y_vl_true, y_vl_pred, average='weighted')\n    })\n\n    # clear for next fold\n    tf.keras.backend.clear_session()\n    gc.collect()\n\n# ── 2) Plot all CV loss curves ────────────────────────────────────────────────\nplt.figure(figsize=(8,5))\nfor fold,(loss,vloss) in loss_curves.items():\n    epochs = range(1, len(loss)+1)\n    plt.plot(epochs, loss,  label=f'Fold {fold} Train', alpha=0.6)\n    plt.plot(epochs, vloss, '--', label=f'Fold {fold} Val',   alpha=0.6)\nplt.xlabel('Epoch'); plt.ylabel('Loss')\nplt.title('Train & Val Loss Curves (All Folds)')\nplt.legend(ncol=2, fontsize='small')\nplt.tight_layout()\nplt.show()\n\n# ── 3) Average CV metrics ────────────────────────────────────────────────────\ndef avg(ms):\n    return {k: np.mean([m[k] for m in ms]) for k in ms[0]}\n\navg_tr = avg(train_metrics)\navg_vl = avg(val_metrics)\n\nprint(\"➡️  Average CV TRAIN metrics:\")\nfor k,v in avg_tr.items():\n    print(f\"   {k}: {v:.3f}\")\nprint(\"➡️  Average CV   VAL metrics:\")\nfor k,v in avg_vl.items():\n    print(f\"   {k}: {v:.3f}\")\n\n# ── 4) Held-out test on best fold ────────────────────────────────────────────\nbest_fold = int(np.argmax([m['F1 Score'] for m in val_metrics]))\nprint(f\"\\n✨ Best fold = {best_fold}\")\n\nmodel = load_model(\n    RESULTS_DIR/f'fold_{best_fold}'/'best_model.keras',\n    custom_objects={'loss_fn': focal_loss(gamma=2.0, alpha=None)}\n)\n\ntest_df = pd.read_csv(TEST_LABEL_CSV)\ntest_df['label'] = (test_df['label'].astype(int)-1).astype(str)\ntest_gen = ImageDataGenerator(rescale=1./255).flow_from_dataframe(\n    test_df,\n    directory=TEST_IMAGE_DIR,\n    x_col='subject', y_col='label',\n    target_size=IMG_SIZE, batch_size=BATCH_SIZE,\n    class_mode='categorical', shuffle=False\n)\n\nprobs  = model.predict(test_gen, verbose=0)\ny_test = test_gen.classes\ny_pred = np.argmax(probs, axis=1)\n\nmetrics_test = {\n    'Accuracy':  accuracy_score(y_test, y_pred),\n    'Precision': precision_score(y_test, y_pred, average='weighted'),\n    'Recall':    recall_score(y_test, y_pred, average='weighted'),\n    'F1 Score':  f1_score(y_test, y_pred, average='weighted'),\n    'Error':     1 - accuracy_score(y_test, y_pred)\n}\n\nprint(\"\\n➡️ Held-out TEST metrics:\")\nfor k,v in metrics_test.items():\n    print(f\"   {k}: {v:.3f}\")\n\n# ── 5) Confusion & ROC on held-out test ───────────────────────────────────────\ncm   = confusion_matrix(y_test, y_pred)\ndisp = ConfusionMatrixDisplay(cm, display_labels=list(test_gen.class_indices))\nplt.figure(figsize=(4,4))\ndisp.plot(ax=plt.gca(), cmap='Blues', colorbar=False)\nplt.title('Held-out Test Confusion')\nplt.show()\n\n# ROC\ny_bin = label_binarize(y_test, classes=list(range(len(test_gen.class_indices))))\nplt.figure(figsize=(6,5))\nfor i,label in enumerate(test_gen.class_indices):\n    fpr, tpr, _ = roc_curve(y_bin[:,i], probs[:,i])\n    plt.plot(fpr, tpr, label=f\"{label} (AUC={auc(fpr,tpr):.2f})\")\nplt.plot([0,1],[0,1],'k--', color='gray')\nplt.xlabel('FPR'); plt.ylabel('TPR')\nplt.title('Held-out Test ROC Curves')\nplt.legend(loc='lower right')\nplt.show()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ─────────────────────────────────────────────────────────────────────────────\n# Cell: Per-Class Precision / Recall / F1 on Held-Out Test Set\n# ─────────────────────────────────────────────────────────────────────────────\n\nimport pandas as pd\nfrom sklearn.metrics import precision_recall_fscore_support, classification_report\n\n# invert the mapping so we get label → class name\ninv_map = {v:k for k,v in test_gen.class_indices.items()}\n# sort by label so class order is consistent\nlabels = sorted(inv_map)\nclass_names = [inv_map[i] for i in labels]\n\n# compute per‐class metrics\nprecisions, recalls, f1s, supports = precision_recall_fscore_support(\n    y_true, y_pred, labels=labels, zero_division=0\n)\n\ndf_metrics = pd.DataFrame({\n    'Class':      class_names,\n    'Support':    supports,\n    'Precision':  precisions,\n    'Recall':     recalls,\n    'F1 Score':   f1s\n})\n\nprint(df_metrics.to_string(index=False))\n\n# print the sklearn report\nprint(\"\\nFull classification report:\\n\")\nprint(classification_report(\n    y_true, y_pred, labels=labels, target_names=class_names, zero_division=0\n))\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Cell: Visualize \n\nimport random\nimport matplotlib.pyplot as plt\nfrom PIL import Image\n\n# how many samples to show\nN = 4\n\n# load the test labels DataFrame (must match the one used above)\ntest_df = pd.read_csv(TEST_LABEL_CSV)\ntest_df['label_int'] = (test_df['label'].astype(int) - 1)\n\n# get class names from your test generator\ninv_map = {v:k for k,v in test_gen.class_indices.items()}\nclasses = [inv_map[i] for i in range(len(inv_map))]\n\n# pick N random rows\nsample_df = test_df.sample(N, random_state=42).reset_index(drop=True)\n\nfig, axes = plt.subplots(1, N, figsize=(4*N, 4))\nfor i, row in sample_df.iterrows():\n    # load image\n    img_path = TEST_IMAGE_DIR / row['subject']\n    img = Image.open(img_path)\n    \n    # find its index in the generator so we can look up the prediction\n    idx = test_gen.filenames.index(row['subject'])\n    pred_idx = y_pred[idx]\n    true_idx = row['label_int']\n    \n    axes[i].imshow(img, cmap='gray')\n    axes[i].axis('off')\n    axes[i].set_title(\n        f\"True: {classes[true_idx]}\\nPred: {classes[pred_idx]}\",\n        fontsize=12\n    )\n\nplt.tight_layout()\nplt.show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ─────────────────────────────────────────────────────────────────────────────\n# Cell: Visualize Sample Cropped Patches\n# ─────────────────────────────────────────────────────────────────────────────\n\nimport os\nimport random\nimport matplotlib.pyplot as plt\nfrom PIL import Image\n\n# Directory containing cropped patches (e.g. fold 0 train)\ncrop_dir = './Neural_Foraminal_Narrowing/fold_0/train'\n\n# List all PNG files\nimg_files = [f for f in os.listdir(crop_dir) if f.lower().endswith('.png')]\nif not img_files:\n    raise RuntimeError(f\"No images found in {crop_dir}\")\n\n# Randomly pick up to 9 samples\nsample_files = random.sample(img_files, min(9, len(img_files)))\n\n# Display in a 3×3 grid\nfig, axes = plt.subplots(3, 3, figsize=(10, 10))\naxes = axes.flatten()\n\nfor ax, fname in zip(axes, sample_files):\n    img = Image.open(os.path.join(crop_dir, fname))\n    ax.imshow(img)\n    ax.set_title(fname, fontsize=8)\n    ax.axis('off')\n\n# Turn off any unused subplots\nfor ax in axes[len(sample_files):]:\n    ax.axis('off')\n\nplt.suptitle('Sample Cropped Patches – Fold 0 Train', fontsize=16)\nplt.tight_layout()\nplt.show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# Cell: Plot Train vs Val Loss per Fold\nimport json\nimport matplotlib.pyplot as plt\nfrom pathlib import Path\n\nfor fold in FOLDS:\n    hist_path = RESULTS_DIR / f'fold_{fold}' / 'history.json'\n    hist = json.load(open(hist_path))\n    epochs = range(1, len(hist['loss']) + 1)\n\n    plt.figure(figsize=(6,4))\n    plt.plot(epochs, hist['loss'],    marker='o', label='Train Loss')\n    plt.plot(epochs, hist['val_loss'],marker='s', label='Val Loss')\n    plt.title(f'Fold {fold} Loss Curve')\n    plt.xlabel('Epoch')\n    plt.ylabel('Loss')\n    plt.legend()\n    plt.tight_layout()\n    plt.show()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}