{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"},"kaggle":{"accelerator":"none","dataSources":[{"sourceType":"competition","sourceId":10200,"databundleVersionId":868375,"isSourceIdPinned":false},{"sourceType":"datasetVersion","sourceId":16215228,"datasetId":10396692,"databundleVersionId":17195548},{"sourceType":"datasetVersion","sourceId":16215217,"datasetId":10396684,"databundleVersionId":17195536}],"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## INSTALLATIONS & IMPORTS","metadata":{}},{"cell_type":"code","source":"!pip install timm scikit-image -q\n\nimport os\nimport re\nimport cv2\nimport copy\nimport pandas as pd\nimport numpy as np\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader\nfrom torchvision import transforms, models\nimport timm\nfrom sklearn.preprocessing import LabelEncoder\nfrom sklearn.metrics import f1_score\nfrom PIL import Image\nimport glob\nfrom skimage.morphology import skeletonize # NEW IMPORT\n\n# Set device to GPU if available\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(f\"Using device: {device}\")","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## PATH CONFIGURATION","metadata":{}},{"cell_type":"code","source":"KAGGLE_INPUT_DIR = '/kaggle/input/datasets/youssefnaggar/bender-dataset-images/images' \nCSV_FILENAME = '/kaggle/input/datasets/youssefnaggar/bender-target-data'\nTRAINING_PATH = '/kaggle/input/datasets/youssefnaggar/bender-dataset-images/images/train'\nVAL_PATH = '/kaggle/input/datasets/youssefnaggar/bender-dataset-images/images/val'\nTEST_PATH = '/kaggle/input/datasets/youssefnaggar/bender-dataset-images/images/test'\n\nCSV_PATH = '/kaggle/input/datasets/youssefnaggar/bender-target-data/Data Model.csv' \nWORKING_DIR = '/kaggle/working'","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## DATA PREPARATION & DATASET","metadata":{}},{"cell_type":"code","source":"def prepare_dataframe(csv_path):\n    df = pd.read_csv(csv_path)\n    df.columns = df.columns.str.strip()\n    \n    label_encoder = LabelEncoder()\n    df['Encoded_Labels'] = label_encoder.fit_transform(df['التصنيفات المرضية'])\n    \n    print(\"\\n--- Clinical Class Mapping ---\")\n    for index, label in enumerate(label_encoder.classes_):\n        print(f\"Class {index}: {label}\")\n        \n    return df, label_encoder\n\ndef preprocess_grayscale(img_path):\n    img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)\n    img = cv2.GaussianBlur(img, (3, 3), 0) # Noise reduction\n    img = cv2.merge([img, img, img])\n    return Image.fromarray(img)\n\ndef preprocess_binarized(img_path):\n    img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)\n    # Adaptive thresholding preserves faint pencil strokes\n    bin_img = cv2.adaptiveThreshold(\n        img, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2\n    )\n    bin_img = cv2.merge([bin_img, bin_img, bin_img])\n    return Image.fromarray(bin_img)\n\ndef preprocess_skeletonized(img_path):\n    # 1. Read in grayscale\n    img = cv2.imread(img_path, cv2.IMREAD_GRAYSCALE)\n    \n    # 2. Binarize (Inverted: strokes become white/255, background becomes black/0)\n    bin_img = cv2.adaptiveThreshold(\n        img, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY_INV, 11, 2\n    )\n    \n    # 3. Convert to boolean array for skimage (True for strokes, False for background)\n    bool_img = bin_img > 127\n    \n    # 4. Apply skeletonization (reduces strokes to 1-pixel width)\n    skeleton = skeletonize(bool_img)\n    \n    # 5. Convert back to uint8 (0 and 255)\n    skeleton_uint8 = (skeleton * 255).astype(np.uint8)\n    \n    # 6. Convert to 3-channel image for the CNN\n    final_img = cv2.merge([skeleton_uint8, skeleton_uint8, skeleton_uint8])\n    return Image.fromarray(final_img)\n\n\nclass BGTCSVDataset(Dataset):\n    def __init__(self, df, img_dir, preprocess_mode='grayscale', transform=None):\n        self.df = df\n        self.img_dir = img_dir\n        self.transform = transform\n        self.preprocess_mode = preprocess_mode\n        self.filepaths = []\n        self.labels = []\n        \n        valid_ext = ('.jpg', '.jpeg', '.png')\n        \n        if not os.path.exists(img_dir):\n            print(f\"Warning: Directory {img_dir} does not exist.\")\n            return\n\n        for file in os.listdir(img_dir):\n            if file.lower().endswith(valid_ext):\n                match = re.search(r'\\d+', file)\n                if match:\n                    case_id = int(match.group())\n                    row = self.df[self.df['رقم الحالة'] == case_id]\n                    \n                    if not row.empty:\n                        label = row['Encoded_Labels'].values[0]\n                        self.filepaths.append(os.path.join(img_dir, file))\n                        self.labels.append(label)\n\n    def __len__(self):\n        return len(self.filepaths)\n\n    def __getitem__(self, idx):\n        img_path = self.filepaths[idx]\n        \n        # Route to the correct preprocessing function\n        if self.preprocess_mode == 'binarized':\n            image = preprocess_binarized(img_path)\n        elif self.preprocess_mode == 'skeletonized':\n            image = preprocess_skeletonized(img_path)\n        else:\n            image = preprocess_grayscale(img_path)\n            \n        label = self.labels[idx]\n        if self.transform:\n            image = self.transform(image)\n        return image, label\n\ntrain_transforms = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.RandomHorizontalFlip(p=0.5),\n    transforms.ColorJitter(brightness=0.2, contrast=0.2),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n])\n\nval_test_transforms = transforms.Compose([\n    transforms.Resize((224, 224)),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n])","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## MODELS","metadata":{}},{"cell_type":"code","source":"def get_resnet50(num_classes=5):\n    model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V1)\n    model.fc = nn.Linear(model.fc.in_features, num_classes)\n    return model\n\ndef get_convnext_v2_tiny(num_classes=5):\n    model = timm.create_model('convnextv2_tiny.fcmae_ft_in1k', pretrained=True, num_classes=num_classes)\n    return model\n\nclass EarlyStopping:\n    def __init__(self, patience=5, min_delta=0):\n        self.patience = patience\n        self.min_delta = min_delta\n        self.counter = 0\n        self.best_loss = None\n        self.early_stop = False\n\n    def __call__(self, val_loss):\n        if self.best_loss is None:\n            self.best_loss = val_loss\n        elif val_loss > self.best_loss - self.min_delta:\n            self.counter += 1\n            if self.counter >= self.patience:\n                self.early_stop = True\n        else:\n            self.best_loss = val_loss\n            self.counter = 0","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## TRAINING & EVALUATION LOGIC","metadata":{}},{"cell_type":"code","source":"def train_model(model, train_loader, val_loader, model_name, num_epochs=30, patience=5, num_classes=5):\n    model = model.to(device)\n    criterion = nn.MSELoss() # Project requirement\n    optimizer = optim.Adam(model.parameters(), lr=1e-4)\n    early_stopping = EarlyStopping(patience=patience)\n    \n    best_model_wts = copy.deepcopy(model.state_dict())\n    best_f1 = 0.0\n\n    for epoch in range(num_epochs):\n        model.train()\n        running_loss = 0.0\n        \n        for inputs, labels in train_loader:\n            inputs = inputs.to(device)\n            # Use the dynamic num_classes parameter here\n            labels_one_hot = nn.functional.one_hot(labels, num_classes=num_classes).float().to(device)\n            \n            optimizer.zero_grad()\n            outputs = model(inputs)\n            outputs_probs = torch.softmax(outputs, dim=1) \n            \n            loss = criterion(outputs_probs, labels_one_hot)\n            loss.backward()\n            optimizer.step()\n            running_loss += loss.item() * inputs.size(0)\n            \n        # Validation Phase\n        model.eval()\n        val_loss = 0.0\n        all_preds, all_labels = [], []\n        \n        with torch.no_grad():\n            for inputs, labels in val_loader:\n                inputs = inputs.to(device)\n                # Use the dynamic num_classes parameter here\n                labels_one_hot = nn.functional.one_hot(labels, num_classes=num_classes).float().to(device)\n                \n                outputs = model(inputs)\n                outputs_probs = torch.softmax(outputs, dim=1)\n                loss = criterion(outputs_probs, labels_one_hot)\n                val_loss += loss.item() * inputs.size(0)\n                \n                _, preds = torch.max(outputs, 1)\n                all_preds.extend(preds.cpu().numpy())\n                all_labels.extend(labels.numpy())\n                \n        epoch_val_loss = val_loss / max(len(val_loader.dataset), 1)\n        epoch_f1 = f1_score(all_labels, all_preds, average='macro') if all_labels else 0\n        \n        print(f\"Epoch {epoch+1} | Val Loss: {epoch_val_loss:.4f} | Val F1: {epoch_f1:.4f}\")\n        \n        if epoch_f1 > best_f1:\n            best_f1 = epoch_f1\n            best_model_wts = copy.deepcopy(model.state_dict())\n            torch.save(best_model_wts, f\"{WORKING_DIR}/best_{model_name}.pth\")\n            \n        early_stopping(epoch_val_loss)\n        if early_stopping.early_stop:\n            print(\"Early stopping triggered!\")\n            break\n            \n    model.load_state_dict(best_model_wts)\n    return model\n\ndef evaluate_model(model, test_loader):\n    model.eval()\n    all_preds, all_labels = [], []\n    \n    with torch.no_grad():\n        for inputs, labels in test_loader:\n            inputs = inputs.to(device)\n            outputs = model(inputs)\n            _, preds = torch.max(outputs, 1)\n            all_preds.extend(preds.cpu().numpy())\n            all_labels.extend(labels.numpy())\n            \n    if not all_labels: return 0.0\n    return f1_score(all_labels, all_preds, average='macro')","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## EXECUTE THE 4 EXPERIMENTS","metadata":{}},{"cell_type":"code","source":"print(\"Loading CSV and encoding labels...\")\ndf, label_encoder = prepare_dataframe(CSV_PATH)\n\nresults = {}\n\n# Added 'skeletonized' experiments to the config list\nconfigs = [\n    ('ResNet50', 'grayscale', get_resnet50),\n    ('ResNet50', 'binarized', get_resnet50),\n    ('ResNet50', 'skeletonized', get_resnet50),\n    ('ConvNeXt_Tiny', 'grayscale', get_convnext_v2_tiny),\n    ('ConvNeXt_Tiny', 'binarized', get_convnext_v2_tiny),\n    ('ConvNeXt_Tiny', 'skeletonized', get_convnext_v2_tiny),\n]\n\nfor model_name, prep_mode, model_fn in configs:\n    experiment_name = f\"{model_name}_{prep_mode}\"\n    print(f\"\\n{'='*50}\\nRunning Experiment: {experiment_name}\\n{'='*50}\")\n    \n    train_data = BGTCSVDataset(df, TRAINING_PATH, preprocess_mode=prep_mode, transform=train_transforms)\n    val_data = BGTCSVDataset(df, VAL_PATH, preprocess_mode=prep_mode, transform=val_test_transforms)\n    test_data = BGTCSVDataset(df, TEST_PATH, preprocess_mode=prep_mode, transform=val_test_transforms)\n    \n    if len(train_data) == 0:\n        print(f\"Skipping {experiment_name}: No training data found in {TRAINING_PATH}.\")\n        continue\n        \n    train_loader = DataLoader(train_data, batch_size=32, shuffle=True)\n    val_loader = DataLoader(val_data, batch_size=32, shuffle=False)\n    test_loader = DataLoader(test_data, batch_size=32, shuffle=False)\n    \n    model = model_fn(num_classes=5)\n    trained_model = train_model(model, train_loader, val_loader, model_name=experiment_name)\n    test_f1 = evaluate_model(trained_model, test_loader)\n    \n    print(f\"\\n>>> Final Test F1 Score for {experiment_name}: {test_f1:.4f} <<<\\n\")\n    results[experiment_name] = test_f1\n\nprint(\"\\n=== FINAL EXPERIMENT COMPARISON ===\")\nfor exp, score in results.items():\n    print(f\"{exp}: {score:.4f}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## QUICKDRAW PRE-TRAINING & FINE-TUNING","metadata":{}},{"cell_type":"code","source":"# 1. PATHS FOR QUICKDRAW\nQUICKDRAW_DIR = '/kaggle/input/quickdraw-doodle-recognition/train_simplified' \n\n# Select 10 geometric/line-based classes relevant to BGT to save time and compute\nQD_CLASSES = ['circle', 'square', 'triangle', 'line', 'hexagon', \n              'octagon', 'diamond', 'zigzag', 'star', 'squiggle']\n\nclass QuickDrawDataset(Dataset):\n    def __init__(self, npy_dir, class_list, transform=None, samples_per_class=2000):\n        self.transform = transform\n        self.images = []\n        self.labels = []\n        \n        print(\"Loading QuickDraw subset...\")\n        for label_idx, cls_name in enumerate(class_list):\n            file_path = os.path.join(npy_dir, f\"{cls_name}.npy\")\n            if os.path.exists(file_path):\n                # Load numpy array (N, 784), take a subset, reshape to (28, 28)\n                data = np.load(file_path)[:samples_per_class]\n                data = data.reshape(-1, 28, 28).astype(np.uint8)\n                \n                for img_array in data:\n                    # Convert to 3-channel image to match our model architecture\n                    img = cv2.merge([img_array, img_array, img_array])\n                    self.images.append(Image.fromarray(img))\n                    self.labels.append(label_idx)\n            else:\n                print(f\"Warning: QuickDraw class '{cls_name}.npy' not found at {file_path}.\")\n                \n        print(f\"Loaded {len(self.images)} QuickDraw images across {len(class_list)} classes.\")\n\n    def __len__(self):\n        return len(self.images)\n\n    def __getitem__(self, idx):\n        img = self.images[idx]\n        label = self.labels[idx]\n        if self.transform:\n            img = self.transform(img)\n        return img, label\n\ndef execute_final_strategy(best_model_name, best_prep_mode, best_original_f1):\n    print(f\"\\n{'='*50}\")\n    print(f\"PHASE 1: PRE-TRAINING ON QUICKDRAW\")\n    print(f\"Using architecture from best experiment: {best_model_name} ({best_prep_mode})\")\n    print(f\"{'='*50}\")\n\n    # A. Setup QuickDraw DataLoader\n    qd_dataset = QuickDrawDataset(QUICKDRAW_DIR, QD_CLASSES, transform=train_transforms)\n    \n    if len(qd_dataset) == 0:\n        print(\"QuickDraw data not found. Please check QUICKDRAW_DIR path. Skipping...\")\n        return\n        \n    train_size = int(0.8 * len(qd_dataset))\n    val_size = len(qd_dataset) - train_size\n    qd_train_data, qd_val_data = torch.utils.data.random_split(qd_dataset, [train_size, val_size])\n        \n    qd_loader = DataLoader(qd_train_data, batch_size=64, shuffle=True)\n    qd_val_loader = DataLoader(qd_val_data, batch_size=64, shuffle=False) \n\n    # B. Initialize Model for QuickDraw (10 classes)\n    if 'ResNet50' in best_model_name:\n        model = get_resnet50(num_classes=len(QD_CLASSES))\n    else:\n        model = get_convnext_v2_tiny(num_classes=len(QD_CLASSES))\n\n    # C. Train on QuickDraw (Pass num_classes = 10)\n    print(\"Training on QuickDraw data...\")\n    model = train_model(model, qd_loader, qd_val_loader, model_name=\"QuickDraw_Pretrained\", \n                        num_epochs=5, patience=2, num_classes=len(QD_CLASSES))\n\n    print(f\"\\n{'='*50}\")\n    print(f\"PHASE 2: FINE-TUNING ON BGT DATASET\")\n    print(f\"{'='*50}\")\n\n    # D. Modify the head for BGT (5 classes)\n    print(\"Replacing classification head for 5 BGT classes...\")\n    if 'ResNet50' in best_model_name:\n        model.fc = nn.Linear(model.fc.in_features, 5)\n    else:\n        model.head.fc = nn.Linear(model.head.fc.in_features, 5)\n\n    # E. Setup BGT DataLoaders using the global defined paths\n    bgt_train_data = BGTCSVDataset(df, TRAINING_PATH, preprocess_mode=best_prep_mode, transform=train_transforms)\n    bgt_val_data = BGTCSVDataset(df, VAL_PATH, preprocess_mode=best_prep_mode, transform=val_test_transforms)\n    bgt_test_data = BGTCSVDataset(df, TEST_PATH, preprocess_mode=best_prep_mode, transform=val_test_transforms)\n    \n    bgt_train_loader = DataLoader(bgt_train_data, batch_size=32, shuffle=True)\n    bgt_val_loader = DataLoader(bgt_val_data, batch_size=32, shuffle=False)\n    bgt_test_loader = DataLoader(bgt_test_data, batch_size=32, shuffle=False)\n\n    # F. Fine-Tune with Lower Learning Rate\n    model = model.to(device)\n    criterion = nn.MSELoss()\n    optimizer = optim.Adam(model.parameters(), lr=1e-5) \n    early_stopping = EarlyStopping(patience=5)\n    \n    best_ft_wts = copy.deepcopy(model.state_dict())\n    best_ft_f1 = 0.0\n\n    print(\"Fine-tuning on BGT dataset...\")\n    for epoch in range(20): \n        model.train()\n        for inputs, labels in bgt_train_loader:\n            inputs = inputs.to(device)\n            labels_one_hot = nn.functional.one_hot(labels, num_classes=5).float().to(device)\n            \n            optimizer.zero_grad()\n            outputs = model(inputs)\n            outputs_probs = torch.softmax(outputs, dim=1) \n            loss = criterion(outputs_probs, labels_one_hot)\n            loss.backward()\n            optimizer.step()\n            \n        model.eval()\n        val_loss = 0.0\n        all_preds, all_labels = [], []\n        \n        with torch.no_grad():\n            for inputs, labels in bgt_val_loader:\n                inputs = inputs.to(device)\n                labels_one_hot = nn.functional.one_hot(labels, num_classes=5).float().to(device)\n                \n                outputs = model(inputs)\n                outputs_probs = torch.softmax(outputs, dim=1)\n                loss = criterion(outputs_probs, labels_one_hot)\n                val_loss += loss.item() * inputs.size(0)\n                \n                _, preds = torch.max(outputs, 1)\n                all_preds.extend(preds.cpu().numpy())\n                all_labels.extend(labels.numpy())\n                \n        epoch_val_loss = val_loss / max(len(bgt_val_loader.dataset), 1)\n        epoch_f1 = f1_score(all_labels, all_preds, average='macro')\n        \n        print(f\"FT Epoch {epoch+1} | Val Loss: {epoch_val_loss:.4f} | Val F1: {epoch_f1:.4f}\")\n        \n        if epoch_f1 > best_ft_f1:\n            best_ft_f1 = epoch_f1\n            best_ft_wts = copy.deepcopy(model.state_dict())\n            torch.save(best_ft_wts, f\"{WORKING_DIR}/Final_QuickDraw_BGT_{best_model_name}.pth\")\n            \n        early_stopping(epoch_val_loss)\n        if early_stopping.early_stop:\n            print(\"Early stopping triggered during fine-tuning!\")\n            break\n\n    # G. Final Evaluation & Comparison\n    model.load_state_dict(best_ft_wts)\n    final_test_f1 = evaluate_model(model, bgt_test_loader)\n    \n    print(f\"\\n{'='*50}\")\n    print(\"FINAL EXPERIMENT RESULTS\")\n    print(f\"{'='*50}\")\n    print(f\"Original Best F1 Score (ImageNet only): {best_original_f1:.4f}\")\n    print(f\"New F1 Score (QuickDraw + ImageNet):    {final_test_f1:.4f}\")\n    \n    if final_test_f1 > best_original_f1:\n        print(\"\\n✅ SUCCESS: Pre-training on hand-drawn shapes improved the model's ability to classify clinical BGT features!\")\n        improvement = ((final_test_f1 - best_original_f1) / best_original_f1) * 100\n        print(f\"Improvement: +{improvement:.2f}%\")\n    else:\n        print(\"\\n⚠️ Note: QuickDraw pre-training did not yield a higher F1 score on the test set. ImageNet weights were sufficient.\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"## RUN THE FINAL PIPELINE","metadata":{}},{"cell_type":"code","source":"# Find the best model from the dictionary created in the experiments cell\nif 'results' in locals() and results:\n    best_experiment = max(results, key=results.get)\n    best_f1_score = results[best_experiment]\n    \n    # Split the string (e.g., \"ConvNeXt_Tiny_skeletonized\") to get the architecture and prep mode\n    if 'ResNet50' in best_experiment:\n        best_arch = 'ResNet50'\n    else:\n        best_arch = 'ConvNeXt_Tiny'\n        \n    # FIX: Added check for skeletonized\n    if 'skeletonized' in best_experiment:\n        best_prep = 'skeletonized'\n    elif 'binarized' in best_experiment:\n        best_prep = 'binarized'\n    else:\n        best_prep = 'grayscale'\n        \n    execute_final_strategy(best_model_name=best_arch, \n                           best_prep_mode=best_prep, \n                           best_original_f1=best_f1_score)\nelse:\n    print(\"Please run the experiment cell first to generate the 'results' dictionary.\")","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}