{"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"}],"dockerImageVersionId":31040,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# !pip install pydicom\n# !pip install matplotlib seaborn scikit-learn","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-02T07:50:14.505277Z","iopub.execute_input":"2025-06-02T07:50:14.505536Z","iopub.status.idle":"2025-06-02T07:50:21.258488Z","shell.execute_reply.started":"2025-06-02T07:50:14.505508Z","shell.execute_reply":"2025-06-02T07:50:21.257783Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# import os\n# import numpy as np\n# import pandas as pd\n# import cv2\n# import pydicom\n# import tensorflow as tf\n# from tensorflow.keras import layers, models, optimizers, callbacks\n# from sklearn.model_selection import train_test_split\n# from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix\n# import matplotlib.pyplot as plt\n# import seaborn as sns\n# from sklearn.utils.class_weight import compute_class_weight","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-02T07:50:21.260060Z","iopub.execute_input":"2025-06-02T07:50:21.260835Z","iopub.status.idle":"2025-06-02T07:50:35.627392Z","shell.execute_reply.started":"2025-06-02T07:50:21.260805Z","shell.execute_reply":"2025-06-02T07:50:35.626640Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"!pip install pydicom\n!pip install matplotlib seaborn scikit-learn","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import os\nimport numpy as np\nimport pandas as pd\nimport cv2\nimport pydicom\nimport tensorflow as tf\nfrom tensorflow.keras import layers, models, optimizers, callbacks\nfrom tensorflow.keras.applications import EfficientNetB1\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix\nfrom sklearn.utils.class_weight import compute_class_weight\nimport matplotlib.pyplot as plt\nimport seaborn as sns","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# !pip install pydicom\n# !pip install matplotlib seaborn scikit-learn\n\n# import os\n# import numpy as np\n# import pandas as pd\n# import cv2\n# import pydicom\n# import tensorflow as tf\n# from tensorflow.keras import layers, models, optimizers, callbacks\n# from tensorflow.keras.applications import EfficientNetB1\n# from sklearn.model_selection import train_test_split\n# from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix\n# from sklearn.utils.class_weight import compute_class_weight\n# import matplotlib.pyplot as plt\n# import seaborn as sns\n\n# Set up mixed precision\ntf.keras.mixed_precision.set_global_policy('mixed_bfloat16')\n\n# Define dataset path\ntrain_path = '/kaggle/input/rsna-2024-lumbar-spine-degenerative-classification/'\n\n# Load CSV files\ntrain = pd.read_csv(os.path.join(train_path, 'train.csv'))\nlabel = pd.read_csv(os.path.join(train_path, 'train_label_coordinates.csv'))\ntrain_desc = pd.read_csv(os.path.join(train_path, 'train_series_descriptions.csv'))\n\n# Reshape train.csv\ndef reshape_row(row):\n    data = {'study_id': [], 'condition': [], 'level': [], 'severity': []}\n    for column, value in row.items():\n        if column != 'study_id':\n            parts = column.split('_')\n            condition = ' '.join([word.capitalize() for word in parts[:-2]])\n            level = parts[-2].capitalize() + '/' + parts[-1].capitalize()\n            data['study_id'].append(row['study_id'])\n            data['condition'].append(condition)\n            data['level'].append(level)\n            data['severity'].append(value)\n    return pd.DataFrame(data)\n\n# Create and merge DataFrames\nnew_train_df = pd.concat([reshape_row(row) for _, row in train.iterrows()], ignore_index=True)\nmerged_df = pd.merge(new_train_df, label, on=['study_id', 'condition', 'level'], how='inner')\nfinal_merged_df = pd.merge(merged_df, train_desc, on=['series_id', 'study_id'], how='inner')\n\n# Create image paths\nfinal_merged_df['image_path'] = (\n    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# Map severity labels\nfinal_merged_df['severity'] = final_merged_df['severity'].map({\n    'Normal/Mild': 'normal_mild',\n    'Moderate': 'moderate',\n    'Severe': 'severe'\n})\n\n# Filter invalid rows\nfinal_merged_df = final_merged_df[final_merged_df['severity'].isin(['normal_mild', 'moderate', 'severe'])]\n\n# Compute class weights\nclass_counts = final_merged_df['severity'].value_counts().sort_index().values\nclass_weights = compute_class_weight('balanced', classes=np.array([0, 1, 2]), \n                                    y=final_merged_df['severity'].map({'normal_mild': 0, 'moderate': 1, 'severe': 2}))\nclass_weights_dict = dict(enumerate(class_weights))\n\n# Define Focal Loss\ndef focal_loss(y_true, y_pred, alpha=list(class_weights_dict.values()), gamma=2.0):\n    y_true = tf.cast(y_true, tf.int32)\n    ce = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y_true, logits=y_pred)\n    probs = tf.nn.softmax(y_pred, axis=-1)\n    probs = tf.gather(probs, y_true, batch_dims=1)\n    alpha = tf.gather(alpha, y_true)\n    modulating_factor = tf.pow(1.0 - probs, gamma)\n    return tf.reduce_mean(alpha * modulating_factor * ce)\n\n# Image preprocessing functions\ndef apply_clahe(image, clip_limit=2, tile_grid_size=(16, 16)):\n    if len(image.shape) == 3 and image.shape[2] == 3:\n        image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n    if image.dtype != np.uint8:\n        image = (image * 255).astype(np.uint8) if np.issubdtype(image.dtype, np.floating) else image.astype(np.uint8)\n    clahe = cv2.createCLAHE(clipLimit=clip_limit, tileGridSize=tile_grid_size)\n    return clahe.apply(image)\n\ndef apply_bilateral_filter(image, diameter=5, sigma_color=10, sigma_space=10):\n    if image.dtype == np.float32 or image.dtype == np.float64:\n        if image.max() <= 1.0:\n            sigma_color = sigma_color / 255.0\n        else:\n            image = (image / 255.0).astype(np.float32)\n    elif image.dtype != np.uint8:\n        image = image.astype(np.uint8)\n    return cv2.bilateralFilter(image, diameter, sigma_color, sigma_space)\n\ndef augment_image(image):\n    image = tf.image.random_brightness(image, max_delta=5/255.0)\n    image = tf.image.random_contrast(image, lower=0.9, upper=1.1)\n    image_uint8 = tf.image.convert_image_dtype(image, tf.uint8)\n    image_uint8 = tf.image.random_jpeg_quality(image_uint8, 80, 100)\n    return tf.image.convert_image_dtype(image_uint8, tf.float32)\n\ndef remove_background(image):\n    if len(image.shape) == 3 and image.shape[2] == 3:\n        image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n    if image.dtype != np.uint8:\n        image = (image * 255).astype(np.uint8) if image.dtype == np.float32 else image.astype(np.uint8)\n    _, mask = cv2.threshold(image, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)\n    return cv2.bitwise_and(image, image, mask=mask)\n\ndef load_dicom_tf(path):\n    dicom = pydicom.dcmread(path.numpy().decode('utf-8'))\n    data = dicom.pixel_array\n    data = data - np.min(data)\n    if np.max(data) != 0:\n        data = data / np.max(data)\n    return data.astype(np.float32)\n\ndef preprocess_image(image, label=None, is_training=False):\n    def process_with_opencv(img_tensor):\n        img_np = img_tensor.numpy().squeeze()\n        img_np = apply_bilateral_filter(img_np)\n        img_np = apply_clahe(img_np)\n        img_np = remove_background(img_np)\n        if is_training:\n            img_np = augment_image(tf.expand_dims(img_np, axis=-1)).numpy().squeeze()\n        img_np = np.expand_dims(img_np, axis=-1)\n        return img_np.astype(np.float32)\n\n    image = tf.py_function(load_dicom_tf, [image], tf.float32)\n    image.set_shape([None, None])\n    image = tf.py_function(process_with_opencv, [image], tf.float32)\n    image.set_shape([None, None, 1])\n    image = tf.image.resize(image, [224, 224])\n    image = tf.image.grayscale_to_rgb(image)\n    image = tf.keras.applications.efficientnet.preprocess_input(image)\n    return (image, label) if label is not None else image\n\n# Create dataset\ndef create_dataset(df, batch_size=32, is_training=False):\n    labels = df['severity'].map({'normal_mild': 0, 'moderate': 1, 'severe': 2}).astype(np.int32)\n    dataset = tf.data.Dataset.from_tensor_slices((df['image_path'], labels))\n    dataset = dataset.map(lambda x, y: preprocess_image(x, y, is_training), num_parallel_calls=tf.data.AUTOTUNE)\n    \n    if is_training:\n        dataset = dataset.apply(tf.data.experimental.rejection_resample(\n            class_func=lambda image, label: label,\n            target_dist=list(class_weights_dict.values()),\n            initial_dist=list(class_weights_dict.values())\n        )).map(lambda resampled_label, original_sample: original_sample)\n    \n    dataset = dataset.batch(batch_size).prefetch(tf.data.AUTOTUNE)\n    return dataset\n\n# Build U-Net encoder\ndef build_unet_encoder(input_shape=(224, 224, 3)):\n    inputs = layers.Input(shape=input_shape)\n    c1 = layers.Conv2D(64, 3, padding='same', activation='relu')(inputs)\n    c1 = layers.Conv2D(64, 3, padding='same', activation='relu')(c1)\n    p1 = layers.MaxPooling2D((2, 2))(c1)\n    \n    c2 = layers.Conv2D(128, 3, padding='same', activation='relu')(p1)\n    c2 = layers.Conv2D(128, 3, padding='same', activation='relu')(c2)\n    p2 = layers.MaxPooling2D((2, 2))(c2)\n    \n    c3 = layers.Conv2D(256, 3, padding='same', activation='relu')(p2)\n    c3 = layers.Conv2D(256, 3, padding='same', activation='relu')(c3)\n    p3 = layers.MaxPooling2D((2, 2))(c3)\n    \n    c4 = layers.Conv2D(512, 3, padding='same', activation='relu')(p3)\n    c4 = layers.Conv2D(512, 3, padding='same', activation='relu')(c4)\n    p4 = layers.MaxPooling2D((2, 2))(c4)\n    \n    c5 = layers.Conv2D(1024, 3, padding='same', activation='relu')(p4)\n    c5 = layers.Conv2D(1024, 3, padding='same', activation='relu')(c5)\n    \n    return models.Model(inputs, c5)\n\n# Build hybrid U-Net + EfficientNetB1 model\ndef build_hybrid_model(input_shape=(224, 224, 3), num_classes=3):\n    # U-Net encoder\n    unet_encoder = build_unet_encoder(input_shape)\n    \n    # EfficientNetB1\n    efficientnet = EfficientNetB1(\n        include_top=False,\n        weights='imagenet',\n        input_shape=input_shape\n    )\n    \n    # Freeze most EfficientNet layers\n    for layer in efficientnet.layers:\n        if 'block6' in layer.name or 'block7' in layer.name:\n            layer.trainable = True\n        else:\n            layer.trainable = False\n    \n    inputs = layers.Input(shape=input_shape)\n    \n    # Get features from U-Net encoder\n    unet_features = unet_encoder(inputs)\n    \n    # Get features from EfficientNetB1\n    eff_features = efficientnet(inputs)\n    \n    # Combine features (concatenate after global average pooling)\n    unet_pooled = layers.GlobalAveragePooling2D()(unet_features)\n    eff_pooled = layers.GlobalAveragePooling2D()(eff_features)\n    combined = layers.Concatenate()([unet_pooled, eff_pooled])\n    \n    # Dense layers for classification\n    x = layers.Dense(512, activation='relu')(combined)\n    x = layers.Dropout(0.5)(x)\n    outputs = layers.Dense(num_classes, activation='linear')(x)\n    \n    return models.Model(inputs, outputs)\n\n# Train model\ndef train_model(model, train_ds, val_ds, series_name):\n    model.compile(optimizer=optimizers.Adam(learning_rate=0.0001), \n                  loss=focal_loss, \n                  metrics=['accuracy'])\n    callbacks_list = [\n        callbacks.EarlyStopping(patience=10, restore_best_weights=True),\n        callbacks.ModelCheckpoint(f'best_{series_name}_hybrid.keras', save_best_only=True)\n    ]\n    history = model.fit(train_ds, validation_data=val_ds, epochs=2, callbacks=callbacks_list)\n    return history\n\n# Evaluate model\ndef evaluate_model(model, test_ds, class_names):\n    y_true, y_pred = [], []\n    for images, labels in test_ds:\n        y_true.extend(labels.numpy())\n        preds = model.predict(images)\n        y_pred.extend(np.argmax(preds, axis=1))\n    y_true, y_pred = np.array(y_true), np.array(y_pred)\n    cm = confusion_matrix(y_true, y_pred)\n    \n    cm_details = {}\n    for i, class_name in enumerate(class_names):\n        tp = cm[i, i]\n        fp = cm[:, i].sum() - tp\n        fn = cm[i, :].sum() - tp\n        tn = cm.sum() - (tp + fp + fn)\n        cm_details[class_name] = {\n            'True Positives': int(tp),\n            'False Positives': int(fp),\n            'False Negatives': int(fn),\n            'True Negatives': int(tn)\n        }\n    \n    return {\n        'precision': precision_score(y_true, y_pred, average='weighted', zero_division=0),\n        'recall': recall_score(y_true, y_pred, average='weighted', zero_division=0),\n        'f1': f1_score(y_true, y_pred, average='weighted', zero_division=0),\n        'cm': cm,\n        'cm_details': cm_details\n    }\n\n# Plot confusion matrix\ndef plot_confusion_matrix(cm, classes, title):\n    plt.figure(figsize=(8, 6))\n    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=classes, yticklabels=classes)\n    plt.title(title)\n    plt.xlabel('Predicted')\n    plt.ylabel('True')\n    file_path = f'{title.lower().replace(\" \", \"_\")}.png'\n    directory = os.path.dirname(file_path)\n    if directory and not os.path.exists(directory):\n        os.makedirs(directory)\n    plt.savefig(file_path)\n    plt.close()\n\n# Main loop for each series\nseries_types = ['Sagittal T1', 'Axial T2', 'Sagittal T2/STIR']\nclass_names = ['normal_mild', 'moderate', 'severe']\n\nfor series_name in series_types:\n    series_df = final_merged_df[final_merged_df['series_description'] == series_name].copy()\n    if series_df.empty:\n        print(f\"No valid data for {series_name}.\")\n        continue\n    \n    # Split data\n    train_df, temp_df = train_test_split(series_df, test_size=0.3, stratify=series_df['severity'], random_state=42)\n    val_df, test_df = train_test_split(temp_df, test_size=0.6667, stratify=temp_df['severity'], random_state=42)\n    \n    print(f\"\\nClass distribution for {series_name}:\")\n    print(\"Train:\", train_df['severity'].value_counts().to_dict())\n    print(\"Validation:\", val_df['severity'].value_counts().to_dict())\n    print(\"Test:\", test_df['severity'].value_counts().to_dict())\n    \n    # Create datasets\n    train_ds = create_dataset(train_df, batch_size=32, is_training=True)\n    val_ds = create_dataset(val_df, batch_size=32)\n    test_ds = create_dataset(test_df, batch_size=32)\n    \n    # Build and train hybrid model\n    model = build_hybrid_model()\n    history = train_model(model, train_ds, val_ds, series_name)\n    \n    # Evaluate\n    results = evaluate_model(model, test_ds, class_names)\n    print(f\"\\nMetrics for {series_name}:\")\n    print(f\"Precision: {results['precision']:.4f}\")\n    print(f\"Recall: {results['recall']:.4f}\")\n    print(f\"F1-Score: {results['f1']:.4f}\")\n    \n    # Print confusion matrix details\n    print(f\"\\nConfusion Matrix Details for {series_name}:\")\n    for class_name, metrics in results['cm_details'].items():\n        print(f\"\\nClass: {class_name}\")\n        for metric_name, value in metrics.items():\n            print(f\"{metric_name}: {value}\")\n            \n    # Plot confusion matrix\n    plot_confusion_matrix(results['cm'], class_names, f'{series_name} Confusion Matrix')\n    \n    # Plot training metrics\n    plt.figure(figsize=(12, 5))\n    plt.subplot(1, 2, 1)\n    plt.plot(history.history['loss'], label='Train Loss')\n    plt.plot(history.history['val_loss'], label='Val Loss')\n    plt.title(f'{series_name} Loss')\n    plt.legend()\n    plt.subplot(1, 2, 2)\n    plt.plot(history.history['accuracy'], label='Train Acc')\n    plt.plot(history.history['val_accuracy'], label='Val Acc')\n    plt.title(f'{series_name} Accuracy')\n    plt.legend()\n    file_path = f'{series_name.lower().replace(\" \", \"_\")}_plots.png'\n    directory = os.path.dirname(file_path)\n    if directory and not os.path.exists(directory):\n        os.makedirs(directory)\n    plt.savefig(file_path)\n    plt.close()","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# !pip install pydicom\n# !pip install matplotlib seaborn scikit-learn\n\n# import os\n# import numpy as np\n# import pandas as pd\n# import cv2\n# import pydicom\n# import tensorflow as tf\n# from tensorflow.keras import layers, models, optimizers, callbacks\n# from sklearn.model_selection import train_test_split\n# from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix\n# import matplotlib.pyplot as plt\n# import seaborn as sns\n# from sklearn.utils.class_weight import compute_class_weight\n# # Set up mixed precision\n# tf.keras.mixed_precision.set_global_policy('mixed_bfloat16')\n\n# # Define dataset path\n# train_path = '/kaggle/input/rsna-2024-lumbar-spine-degenerative-classification/'\n\n# # Load CSV files\n# train = pd.read_csv(os.path.join(train_path, 'train.csv'))\n# label = pd.read_csv(os.path.join(train_path, 'train_label_coordinates.csv'))\n# train_desc = pd.read_csv(os.path.join(train_path, 'train_series_descriptions.csv'))\n\n# # Reshape train.csv\n# def reshape_row(row):\n#     data = {'study_id': [], 'condition': [], 'level': [], 'severity': []}\n#     for column, value in row.items():\n#         if column != 'study_id':\n#             parts = column.split('_')\n#             condition = ' '.join([word.capitalize() for word in parts[:-2]])\n#             level = parts[-2].capitalize() + '/' + parts[-1].capitalize()\n#             data['study_id'].append(row['study_id'])\n#             data['condition'].append(condition)\n#             data['level'].append(level)\n#             data['severity'].append(value)\n#     return pd.DataFrame(data)\n\n# # Create and merge DataFrames\n# new_train_df = pd.concat([reshape_row(row) for _, row in train.iterrows()], ignore_index=True)\n# merged_df = pd.merge(new_train_df, label, on=['study_id', 'condition', 'level'], how='inner')\n# final_merged_df = pd.merge(merged_df, train_desc, on=['series_id', 'study_id'], how='inner')\n\n# # Create image paths\n# final_merged_df['image_path'] = (\n#     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# # Map severity labels\n# final_merged_df['severity'] = final_merged_df['severity'].map({\n#     'Normal/Mild': 'normal_mild',\n#     'Moderate': 'moderate',\n#     'Severe': 'severe'\n# })\n\n# # Filter invalid rows\n# final_merged_df = final_merged_df[final_merged_df['severity'].isin(['normal_mild', 'moderate', 'severe'])]\n\n# # Compute class weights\n# class_counts = final_merged_df['severity'].value_counts().sort_index().values\n# class_weights = compute_class_weight('balanced', classes=np.array([0, 1, 2]), \n#                                     y=final_merged_df['severity'].map({'normal_mild': 0, 'moderate': 1, 'severe': 2}))\n# class_weights_dict = dict(enumerate(class_weights))\n\n# # Define Focal Loss\n# def focal_loss(y_true, y_pred, alpha=list(class_weights_dict.values()), gamma=2.0):\n#     y_true = tf.cast(y_true, tf.int32)\n#     ce = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y_true, logits=y_pred)\n#     probs = tf.nn.softmax(y_pred, axis=-1)\n#     probs = tf.gather(probs, y_true, batch_dims=1)\n#     alpha = tf.gather(alpha, y_true)\n#     modulating_factor = tf.pow(1.0 - probs, gamma)\n#     return tf.reduce_mean(alpha * modulating_factor * ce)\n\n# # Image preprocessing functions\n# def apply_clahe(image, clip_limit=2, tile_grid_size=(16, 16)):\n#     if len(image.shape) == 3 and image.shape[2] == 3:\n#         image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n#     if image.dtype != np.uint8:\n#         image = (image * 255).astype(np.uint8) if np.issubdtype(image.dtype, np.floating) else image.astype(np.uint8)\n#     clahe = cv2.createCLAHE(clipLimit=clip_limit, tileGridSize=tile_grid_size)\n#     return clahe.apply(image)\n\n# def apply_bilateral_filter(image, diameter=5, sigma_color=10, sigma_space=10):\n#     if image.dtype == np.float32 or image.dtype == np.float64:\n#         if image.max() <= 1.0:\n#             sigma_color = sigma_color / 255.0\n#         else:\n#             image = (image / 255.0).astype(np.float32)\n#     elif image.dtype != np.uint8:\n#         image = image.astype(np.uint8)\n#     return cv2.bilateralFilter(image, diameter, sigma_color, sigma_space)\n\n# def augment_image(image):\n#     image = tf.image.random_brightness(image, max_delta=5/255.0)\n#     image = tf.image.random_contrast(image, lower=0.9, upper=1.1)\n#     image_uint8 = tf.image.convert_image_dtype(image, tf.uint8)\n#     image_uint8 = tf.image.random_jpeg_quality(image_uint8, 80, 100)\n#     return tf.image.convert_image_dtype(image_uint8, tf.float32)\n\n# def remove_background(image):\n#     if len(image.shape) == 3 and image.shape[2] == 3:\n#         image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n#     if image.dtype != np.uint8:\n#         image = (image * 255).astype(np.uint8) if image.dtype == np.float32 else image.astype(np.uint8)\n#     _, mask = cv2.threshold(image, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)\n#     return cv2.bitwise_and(image, image, mask=mask)\n\n# def load_dicom_tf(path):\n#     dicom = pydicom.dcmread(path.numpy().decode('utf-8'))\n#     data = dicom.pixel_array\n#     data = data - np.min(data)\n#     if np.max(data) != 0:\n#         data = data / np.max(data)\n#     return data.astype(np.float32)\n\n# def preprocess_image(image, label=None, is_training=False):\n#     def process_with_opencv(img_tensor):\n#         img_np = img_tensor.numpy().squeeze()\n#         img_np = apply_bilateral_filter(img_np)\n#         img_np = apply_clahe(img_np)\n#         img_np = remove_background(img_np)\n#         if is_training:\n#             img_np = augment_image(tf.expand_dims(img_np, axis=-1)).numpy().squeeze()\n#         img_np = np.expand_dims(img_np, axis=-1)\n#         return img_np.astype(np.float32)\n\n#     image = tf.py_function(load_dicom_tf, [image], tf.float32)\n#     image.set_shape([None, None])\n#     image = tf.py_function(process_with_opencv, [image], tf.float32)\n#     image.set_shape([None, None, 1])\n#     image = tf.image.resize(image, [224, 224])\n#     image = tf.image.grayscale_to_rgb(image)\n#     image = tf.keras.applications.resnet50.preprocess_input(image)\n#     return (image, label) if label is not None else image\n\n# # Create dataset\n# def create_dataset(df, batch_size=32, is_training=False):\n#     labels = df['severity'].map({'normal_mild': 0, 'moderate': 1, 'severe': 2}).astype(np.int32)\n#     dataset = tf.data.Dataset.from_tensor_slices((df['image_path'], labels))\n#     dataset = dataset.map(lambda x, y: preprocess_image(x, y, is_training), num_parallel_calls=tf.data.AUTOTUNE)\n    \n#     if is_training:\n#         dataset = dataset.apply(tf.data.experimental.rejection_resample(\n#             class_func=lambda image, label: label,\n#             target_dist=list(class_weights_dict.values()),\n#             initial_dist=list(class_weights_dict.values())\n#         )).map(lambda resampled_label, original_sample: original_sample)\n    \n#     dataset = dataset.batch(batch_size).prefetch(tf.data.AUTOTUNE)\n#     return dataset\n\n# # Build U-Net encoder\n# def build_unet_encoder(input_shape=(224, 224, 3), num_classes=3):\n#     inputs = layers.Input(shape=input_shape)\n#     c1 = layers.Conv2D(64, 3, padding='same', activation='relu')(inputs)\n#     c1 = layers.Conv2D(64, 3, padding='same', activation='relu')(c1)\n#     p1 = layers.MaxPooling2D((2, 2))(c1)\n    \n#     c2 = layers.Conv2D(128, 3, padding='same', activation='relu')(p1)\n#     c2 = layers.Conv2D(128, 3, padding='same', activation='relu')(c2)\n#     p2 = layers.MaxPooling2D((2, 2))(c2)\n    \n#     c3 = layers.Conv2D(256, 3, padding='same', activation='relu')(p2)\n#     c3 = layers.Conv2D(256, 3, padding='same', activation='relu')(c3)\n#     p3 = layers.MaxPooling2D((2, 2))(c3)\n    \n#     c4 = layers.Conv2D(512, 3, padding='same', activation='relu')(p3)\n#     c4 = layers.Conv2D(512, 3, padding='same', activation='relu')(c4)\n#     p4 = layers.MaxPooling2D((2, 2))(c4)\n    \n#     c5 = layers.Conv2D(1024, 3, padding='same', activation='relu')(p4)\n#     c5 = layers.Conv2D(1024, 3, padding='same', activation='relu')(c5)\n    \n#     x = layers.GlobalAveragePooling2D()(c5)\n#     x = layers.Dense(512, activation='relu')(x)\n#     x = layers.Dropout(0.5)(x)\n#     outputs = layers.Dense(num_classes, activation='linear')(x)\n    \n#     return models.Model(inputs, outputs)\n\n# # Train model\n# def train_model(model, train_ds, val_ds, series_name):\n#     model.compile(optimizer=optimizers.Adam(learning_rate=0.0001), \n#                   loss=focal_loss, \n#                   metrics=['accuracy'])\n#     callbacks_list = [\n#         callbacks.EarlyStopping(patience=10, restore_best_weights=True),\n#         callbacks.ModelCheckpoint(f'best_{series_name}_unet.keras', save_best_only=True)\n#     ]\n#     history = model.fit(train_ds, validation_data=val_ds, epochs=2, callbacks=callbacks_list)\n#     return history\n\n# # Evaluate model and extract confusion matrix details\n# def evaluate_model(model, test_ds, class_names):\n#     y_true, y_pred = [], []\n#     for images, labels in test_ds:\n#         y_true.extend(labels.numpy())\n#         preds = model.predict(images)\n#         y_pred.extend(np.argmax(preds, axis=1))\n#     y_true, y_pred = np.array(y_true), np.array(y_pred)\n#     cm = confusion_matrix(y_true, y_pred)\n    \n#     # Calculate per-class metrics\n#     cm_details = {}\n#     for i, class_name in enumerate(class_names):\n#         tp = cm[i, i]\n#         fp = cm[:, i].sum() - tp\n#         fn = cm[i, :].sum() - tp\n#         tn = cm.sum() - (tp + fp + fn)\n#         cm_details[class_name] = {\n#             'True Positives': int(tp),\n#             'False Positives': int(fp),\n#             'False Negatives': int(fn),\n#             'True Negatives': int(tn)\n#         }\n    \n#     return {\n#         'precision': precision_score(y_true, y_pred, average='weighted', zero_division=0),\n#         'recall': recall_score(y_true, y_pred, average='weighted', zero_division=0),\n#         'f1': f1_score(y_true, y_pred, average='weighted', zero_division=0),\n#         'cm': cm,\n#         'cm_details': cm_details\n#     }\n\n# # Plot confusion matrix\n# def plot_confusion_matrix(cm, classes, title):\n#     plt.figure(figsize=(8, 6))\n#     sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=classes, yticklabels=classes)\n#     plt.title(title)\n#     plt.xlabel('Predicted')\n#     plt.ylabel('True')\n#     file_path = f'{title.lower().replace(\" \", \"_\")}.png'\n#     directory = os.path.dirname(file_path)\n#     if directory and not os.path.exists(directory):\n#         os.makedirs(directory)\n#     plt.savefig(file_path)\n#     plt.close()\n\n# # Main loop for each series\n# series_types = ['Sagittal T1', 'Axial T2', 'Sagittal T2/STIR']\n# class_names = ['normal_mild', 'moderate', 'severe']\n\n# for series_name in series_types:\n#     series_df = final_merged_df[final_merged_df['series_description'] == series_name].copy()\n#     if series_df.empty:\n#         print(f\"No valid data for {series_name}.\")\n#         continue\n    \n#     # Split data\n#     train_df, temp_df = train_test_split(series_df, test_size=0.3, stratify=series_df['severity'], random_state=42)\n#     val_df, test_df = train_test_split(temp_df, test_size=0.6667, stratify=temp_df['severity'], random_state=42)\n    \n#     print(f\"\\nClass distribution for {series_name}:\")\n#     print(\"Train:\", train_df['severity'].value_counts().to_dict())\n#     print(\"Validation:\", val_df['severity'].value_counts().to_dict())\n#     print(\"Test:\", test_df['severity'].value_counts().to_dict())\n    \n#     # Create datasets\n#     train_ds = create_dataset(train_df, batch_size=32, is_training=True)\n#     val_ds = create_dataset(val_df, batch_size=32)\n#     test_ds = create_dataset(test_df, batch_size=32)\n    \n#     # Build and train model\n#     model = build_unet_encoder()\n#     history = train_model(model, train_ds, val_ds, series_name)\n    \n#     # Evaluate\n#     results = evaluate_model(model, test_ds, class_names)\n#     print(f\"\\nMetrics for {series_name}:\")\n#     print(f\"Precision: {results['precision']:.4f}\")\n#     print(f\"Recall: {results['recall']:.4f}\")\n#     print(f\"F1-Score: {results['f1']:.4f}\")\n    \n#     # Print confusion matrix details\n#     print(f\"\\nConfusion Matrix Details for {series_name}:\")\n#     for class_name, metrics in results['cm_details'].items():\n#         print(f\"\\nClass: {class_name}\")\n#         for metric_name, value in metrics.items():\n#             print(f\"{metric_name}: {value}\")\n            \n#     # Plot confusion matrix\n#     plot_confusion_matrix(results['cm'], class_names, f'{series_name} Confusion Matrix')\n    \n#     # Plot training metrics\n#     plt.figure(figsize=(12, 5))\n#     plt.subplot(1, 2, 1)\n#     plt.plot(history.history['loss'], label='Train Loss')\n#     plt.plot(history.history['val_loss'], label='Val Loss')\n#     plt.title(f'{series_name} Loss')\n#     plt.legend()\n#     plt.subplot(1, 2, 2)\n#     plt.plot(history.history['accuracy'], label='Train Acc')\n#     plt.plot(history.history['val_accuracy'], label='Val Acc')\n#     plt.title(f'{series_name} Accuracy')\n#     plt.legend()\n#     file_path = f'{series_name.lower().replace(\" \", \"_\")}_plots.png'\n#     directory = os.path.dirname(file_path)\n#     if directory and not os.path.exists(directory):\n#         os.makedirs(directory)\n#     plt.savefig(file_path)\n#     plt.close()\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# !pip install pydicom\n# !pip install matplotlib seaborn scikit-learn\n# import os\n# import numpy as np\n# import pandas as pd\n# import cv2\n# import pydicom\n# import tensorflow as tf\n# from tensorflow.keras import layers, models, optimizers, callbacks\n# from sklearn.model_selection import train_test_split\n# from sklearn.metrics import precision_score, recall_score, f1_score, confusion_matrix\n# import matplotlib.pyplot as plt\n# import seaborn as sns\n# from sklearn.utils.class_weight import compute_class_weight\n# # Set up mixed precision for better performance\n# tf.keras.mixed_precision.set_global_policy('mixed_bfloat16')\n\n# # Define dataset path\n# train_path = '/kaggle/input/rsna-2024-lumbar-spine-degenerative-classification/'\n\n# # Load CSV files\n# train = pd.read_csv(os.path.join(train_path, 'train.csv'))\n# label = pd.read_csv(os.path.join(train_path, 'train_label_coordinates.csv'))\n# train_desc = pd.read_csv(os.path.join(train_path, 'train_series_descriptions.csv'))\n\n# # Reshape train.csv to separate conditions, levels, and severities\n# def reshape_row(row):\n#     data = {'study_id': [], 'condition': [], 'level': [], 'severity': []}\n#     for column, value in row.items():\n#         if column != 'study_id':\n#             parts = column.split('_')\n#             condition = ' '.join([word.capitalize() for word in parts[:-2]])\n#             level = parts[-2].capitalize() + '/' + parts[-1].capitalize()\n#             data['study_id'].append(row['study_id'])\n#             data['condition'].append(condition)\n#             data['level'].append(level)\n#             data['severity'].append(value)\n#     return pd.DataFrame(data)\n\n# # Create and merge DataFrames\n# new_train_df = pd.concat([reshape_row(row) for _, row in train.iterrows()], ignore_index=True)\n# merged_df = pd.merge(new_train_df, label, on=['study_id', 'condition', 'level'], how='inner')\n# final_merged_df = pd.merge(merged_df, train_desc, on=['series_id', 'study_id'], how='inner')\n\n# # Create image paths\n# final_merged_df['image_path'] = (\n#     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# # Map severity labels\n# final_merged_df['severity'] = final_merged_df['severity'].map({\n#     'Normal/Mild': 'normal_mild',\n#     'Moderate': 'moderate',\n#     'Severe': 'severe'\n# })\n\n# # Filter out invalid rows\n# final_merged_df = final_merged_df[final_merged_df['severity'].isin(['normal_mild', 'moderate', 'severe'])]\n\n# # Compute class weights for imbalanced classes\n# class_counts = final_merged_df['severity'].value_counts().sort_index().values\n# class_weights = compute_class_weight('balanced', classes=np.array([0, 1, 2]), \n#                                     y=final_merged_df['severity'].map({'normal_mild': 0, 'moderate': 1, 'severe': 2}))\n# class_weights_dict = dict(enumerate(class_weights))\n\n# # Define Focal Loss\n# def focal_loss(y_true, y_pred, alpha=list(class_weights_dict.values()), gamma=2.0):\n#     y_true = tf.cast(y_true, tf.int32)\n#     ce = tf.nn.sparse_softmax_cross_entropy_with_logits(labels=y_true, logits=y_pred)\n#     probs = tf.nn.softmax(y_pred, axis=-1)\n#     probs = tf.gather(probs, y_true, batch_dims=1)\n#     alpha = tf.gather(alpha, y_true)\n#     modulating_factor = tf.pow(1.0 - probs, gamma)\n#     return tf.reduce_mean(alpha * modulating_factor * ce)\n\n# # Image preprocessing functions\n# def apply_clahe(image, clip_limit=2, tile_grid_size=(16, 16)):\n#     if len(image.shape) == 3 and image.shape[2] == 3:\n#         image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n#     if image.dtype != np.uint8:\n#         image = (image * 255).astype(np.uint8) if np.issubdtype(image.dtype, np.floating) else image.astype(np.uint8)\n#     clahe = cv2.createCLAHE(clipLimit=clip_limit, tileGridSize=tile_grid_size)\n#     return clahe.apply(image)\n\n# def apply_bilateral_filter(image, diameter=5, sigma_color=10, sigma_space=10):\n#     if image.dtype == np.float32 or image.dtype == np.float64:\n#         if image.max() <= 1.0:\n#             sigma_color = sigma_color / 255.0\n#         else:\n#             image = (image / 255.0).astype(np.float32)\n#     elif image.dtype != np.uint8:\n#         image = image.astype(np.uint8)\n#     return cv2.bilateralFilter(image, diameter, sigma_color, sigma_space)\n\n# def augment_image(image):\n#     image = tf.image.random_brightness(image, max_delta=5/255.0)\n#     image = tf.image.random_contrast(image, lower=0.9, upper=1.1)\n#     image_uint8 = tf.image.convert_image_dtype(image, tf.uint8)\n#     image_uint8 = tf.image.random_jpeg_quality(image_uint8, 80, 100)\n#     return tf.image.convert_image_dtype(image_uint8, tf.float32)\n\n# def remove_background(image):\n#     if len(image.shape) == 3 and image.shape[2] == 3:\n#         image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)\n#     if image.dtype != np.uint8:\n#         image = (image * 255).astype(np.uint8) if image.dtype == np.float32 else image.astype(np.uint8)\n#     _, mask = cv2.threshold(image, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)\n#     return cv2.bitwise_and(image, image, mask=mask)\n\n# def load_dicom_tf(path):\n#     dicom = pydicom.dcmread(path.numpy().decode('utf-8'))\n#     data = dicom.pixel_array\n#     data = data - np.min(data)\n#     if np.max(data) != 0:\n#         data = data / np.max(data)\n#     return data.astype(np.float32)\n\n# def preprocess_image(image, label=None, is_training=False):\n#     def process_with_opencv(img_tensor):\n#         img_np = img_tensor.numpy().squeeze()\n#         img_np = apply_bilateral_filter(img_np)\n#         img_np = apply_clahe(img_np)\n#         img_np = remove_background(img_np)\n#         if is_training:\n#             img_np = augment_image(tf.expand_dims(img_np, axis=-1)).numpy().squeeze()\n#         img_np = np.expand_dims(img_np, axis=-1)\n#         return img_np.astype(np.float32)\n\n#     image = tf.py_function(load_dicom_tf, [image], tf.float32)\n#     image.set_shape([None, None])\n#     image = tf.py_function(process_with_opencv, [image], tf.float32)\n#     image.set_shape([None, None, 1])\n#     image = tf.image.resize(image, [224, 224])\n#     image = tf.image.grayscale_to_rgb(image)\n#     image = tf.keras.applications.resnet50.preprocess_input(image)\n#     return (image, label) if label is not None else image\n\n# # Create dataset\n# def create_dataset(df, batch_size=32, is_training=False):\n#     labels = df['severity'].map({'normal_mild': 0, 'moderate': 1, 'severe': 2}).astype(np.int32)\n#     dataset = tf.data.Dataset.from_tensor_slices((df['image_path'], labels))\n#     dataset = dataset.map(lambda x, y: preprocess_image(x, y, is_training), num_parallel_calls=tf.data.AUTOTUNE)\n    \n#     if is_training:\n#         dataset = dataset.apply(tf.data.experimental.rejection_resample(\n#             class_func=lambda image, label: label,\n#             target_dist=list(class_weights_dict.values()),\n#             initial_dist=list(class_weights_dict.values())\n#         )).map(lambda resampled_label, original_sample: original_sample)\n    \n#     dataset = dataset.batch(batch_size).prefetch(tf.data.AUTOTUNE)\n#     return dataset\n\n# # Build U-Net encoder\n# def build_unet_encoder(input_shape=(224, 224, 3), num_classes=3):\n#     inputs = layers.Input(shape=input_shape)\n#     c1 = layers.Conv2D(64, 3, padding='same', activation='relu')(inputs)\n#     c1 = layers.Conv2D(64, 3, padding='same', activation='relu')(c1)\n#     p1 = layers.MaxPooling2D((2, 2))(c1)\n    \n#     c2 = layers.Conv2D(128, 3, padding='same', activation='relu')(p1)\n#     c2 = layers.Conv2D(128, 3, padding='same', activation='relu')(c2)\n#     p2 = layers.MaxPooling2D((2, 2))(c2)\n    \n#     c3 = layers.Conv2D(256, 3, padding='same', activation='relu')(p2)\n#     c3 = layers.Conv2D(256, 3, padding='same', activation='relu')(c3)\n#     p3 = layers.MaxPooling2D((2, 2))(c3)\n    \n#     c4 = layers.Conv2D(512, 3, padding='same', activation='relu')(p3)\n#     c4 = layers.Conv2D(512, 3, padding='same', activation='relu')(c4)\n#     p4 = layers.MaxPooling2D((2, 2))(c4)\n    \n#     c5 = layers.Conv2D(1024, 3, padding='same', activation='relu')(p4)\n#     c5 = layers.Conv2D(1024, 3, padding='same', activation='relu')(c5)\n    \n#     x = layers.GlobalAveragePooling2D()(c5)\n#     x = layers.Dense(512, activation='relu')(x)\n#     x = layers.Dropout(0.5)(x)\n#     outputs = layers.Dense(num_classes, activation='linear')(x)\n    \n#     return models.Model(inputs, outputs)\n\n# # Train model\n# def train_model(model, train_ds, val_ds, series_name):\n#     model.compile(optimizer=optimizers.Adam(learning_rate=0.0001), \n#                   loss=focal_loss, \n#                   metrics=['accuracy'])\n#     callbacks_list = [\n#         callbacks.EarlyStopping(patience=10, restore_best_weights=True),\n#         callbacks.ModelCheckpoint(f'best_{series_name}_unet.keras', save_best_only=True)\n#     ]\n#     history = model.fit(train_ds, validation_data=val_ds, epochs=2, callbacks=callbacks_list)\n#     return history\n\n# # Evaluate model\n# def evaluate_model(model, test_ds):\n#     y_true, y_pred = [], []\n#     for images, labels in test_ds:\n#         y_true.extend(labels.numpy())\n#         preds = model.predict(images)\n#         y_pred.extend(np.argmax(preds, axis=1))\n#     y_true, y_pred = np.array(y_true), np.array(y_pred)\n#     return {\n#         'precision': precision_score(y_true, y_pred, average='weighted', zero_division=0),\n#         'recall': recall_score(y_true, y_pred, average='weighted', zero_division=0),\n#         'f1': f1_score(y_true, y_pred, average='weighted', zero_division=0),\n#         'cm': confusion_matrix(y_true, y_pred)\n#     }\n\n# # Plot confusion matrix\n# def plot_confusion_matrix(cm, classes, title):\n#     plt.figure(figsize=(8, 6))\n#     sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', xticklabels=classes, yticklabels=classes)\n#     plt.title(title)\n#     plt.xlabel('Predicted')\n#     plt.ylabel('True')\n#     file_path = f'{title.lower().replace(\" \", \"_\")}.png'\n#     directory = os.path.dirname(file_path)\n#     if directory and not os.path.exists(directory):\n#         os.makedirs(directory)\n#     plt.savefig(file_path)\n#     plt.close()\n\n# # Main loop for each series\n# series_types = ['Sagittal T1', 'Axial T2', 'Sagittal T2/STIR']\n# class_names = ['normal_mild', 'moderate', 'severe']\n\n# for series_name in series_types:\n#     series_df = final_merged_df[final_merged_df['series_description'] == series_name].copy()\n#     if series_df.empty:\n#         print(f\"No valid data for {series_name}.\")\n#         continue\n    \n#     # Split data\n#     train_df, temp_df = train_test_split(series_df, test_size=0.3, stratify=series_df['severity'], random_state=42)\n#     val_df, test_df = train_test_split(temp_df, test_size=0.6667, stratify=temp_df['severity'], random_state=42)\n    \n#     print(f\"\\nClass distribution for {series_name}:\")\n#     print(\"Train:\", train_df['severity'].value_counts().to_dict())\n#     print(\"Validation:\", val_df['severity'].value_counts().to_dict())\n#     print(\"Test:\", test_df['severity'].value_counts().to_dict())\n    \n#     # Create datasets\n#     train_ds = create_dataset(train_df, batch_size=32, is_training=True)\n#     val_ds = create_dataset(val_df, batch_size=32)\n#     test_ds = create_dataset(test_df, batch_size=32)\n    \n#     # Build and train model\n#     model = build_unet_encoder()\n#     history = train_model(model, train_ds, val_ds, series_name)\n    \n#     # Evaluate\n#     results = evaluate_model(model, test_ds)\n#     print(f\"\\nMetrics for {series_name}:\")\n#     print(f\"Precision: {results['precision']:.4f}\")\n#     print(f\"Recall: {results['recall']:.4f}\")\n#     print(f\"F1-Score: {results['f1']:.4f}\")\n\n\n#     # Plot confusion matrix\n#     plot_confusion_matrix(results['cm'], class_names, f'{series_name} Confusion Matrix')\n    \n#     # Plot training metrics\n#     plt.figure(figsize=(12, 5))\n#     plt.subplot(1, 2, 1)\n#     plt.plot(history.history['loss'], label='Train Loss')\n#     plt.plot(history.history['val_loss'], label='Val Loss')\n#     plt.title(f'{series_name} Loss')\n#     plt.legend()\n#     plt.subplot(1, 2, 2)\n#     plt.plot(history.history['accuracy'], label='Train Acc')\n#     plt.plot(history.history['val_accuracy'], label='Val Acc')\n#     plt.title(f'{series_name} Accuracy')\n#     plt.legend()\n#     file_path = f'{series_name.lower().replace(\" \", \"_\")}_plots.png'\n#     directory = os.path.dirname(file_path)\n#     if directory and not os.path.exists(directory):\n#         os.makedirs(directory)\n#     plt.savefig(file_path)\n#     plt.close()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-06-02T07:50:35.628153Z","iopub.execute_input":"2025-06-02T07:50:35.628724Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}