{"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":106372,"databundleVersionId":12864984,"sourceType":"competition"}],"dockerImageVersionId":31041,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# This Python 3 environment comes with many helpful analytics libraries installed\n# It is defined by the kaggle/python Docker image: https://github.com/kaggle/docker-python\n# For example, here's several helpful packages to load\n\nimport numpy as np # linear algebra\nimport pandas as pd # data processing, CSV file I/O (e.g. pd.read_csv)\n\n# Input data files are available in the read-only \"../input/\" directory\n# For example, running this (by clicking run or pressing Shift+Enter) will list all files under the input directory\n\nimport os\nfor dirname, _, filenames in os.walk('/kaggle/input'):\n    for filename in filenames:\n        print(os.path.join(dirname, filename))\n\n# You can write up to 20GB to the current directory (/kaggle/working/) that gets preserved as output when you create a version using \"Save & Run All\" \n# You can also write temporary files to /kaggle/temp/, but they won't be saved outside of the current session","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-07-07T04:23:11.906045Z","iopub.execute_input":"2025-07-07T04:23:11.906242Z","iopub.status.idle":"2025-07-07T04:23:28.399158Z","shell.execute_reply.started":"2025-07-07T04:23:11.906214Z","shell.execute_reply":"2025-07-07T04:23:28.398372Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nimport os\nimport cv2\nimport matplotlib.pyplot as plt\nimport seaborn as sns\nfrom sklearn.model_selection import train_test_split, StratifiedKFold\nfrom sklearn.preprocessing import LabelEncoder\nfrom sklearn.metrics import accuracy_score, classification_report, confusion_matrix\nimport torch\nimport torch.nn as nn\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader\nimport torchvision.transforms as transforms\nimport torchvision.models as models\nfrom PIL import Image\nimport warnings\nwarnings.filterwarnings('ignore')\n\n# Set random seeds for reproducibility\ndef set_seed(seed=42):\n    np.random.seed(seed)\n    torch.manual_seed(seed)\n    torch.cuda.manual_seed(seed)\n    torch.cuda.manual_seed_all(seed)\n    torch.backends.cudnn.deterministic = True\n    torch.backends.cudnn.benchmark = False\n\nset_seed(42)\n\n# Device configuration\ndevice = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nprint(f\"Using device: {device}\")\n\n# Data paths\nBASE_DIR = '/kaggle/input/2025-bamboo-summer-competiton-dl-pr'\nTRAIN_DIR = os.path.join(BASE_DIR, 'train')\nTEST_DIR = os.path.join(BASE_DIR, 'test')\nTRAIN_CSV = os.path.join(BASE_DIR, 'train.csv')\nTEST_CSV = os.path.join(BASE_DIR, 'test.csv')\nSAMPLE_SUBMISSION = os.path.join(BASE_DIR, 'sample_submission.csv')\n\n# Load data\ntrain_df = pd.read_csv(TRAIN_CSV)\ntest_df = pd.read_csv(TEST_CSV)\nsample_submission = pd.read_csv(SAMPLE_SUBMISSION)\n\nprint(f\"Training data shape: {train_df.shape}\")\nprint(f\"Test data shape: {test_df.shape}\")\nprint(f\"Number of classes: {train_df['label'].nunique()}\")\n\n# Display class distribution\nclass_counts = train_df['label'].value_counts()\nprint(\"\\nClass distribution:\")\nprint(class_counts)\n\n# Visualize class distribution\nplt.figure(figsize=(12, 6))\nsns.countplot(data=train_df, x='label', order=class_counts.index)\nplt.title('Distribution of Butterfly Species')\nplt.xticks(rotation=45)\nplt.tight_layout()\nplt.show()\n\n# Label encoding\nlabel_encoder = LabelEncoder()\ntrain_df['label_encoded'] = label_encoder.fit_transform(train_df['label'])\nnum_classes = len(label_encoder.classes_)\nprint(f\"Number of classes: {num_classes}\")\n\n# Custom Dataset class\nclass ButterflyDataset(Dataset):\n    def __init__(self, dataframe, root_dir, transform=None, is_test=False):\n        self.dataframe = dataframe\n        self.root_dir = root_dir\n        self.transform = transform\n        self.is_test = is_test\n        \n    def __len__(self):\n        return len(self.dataframe)\n    \n    def __getitem__(self, idx):\n        img_name = self.dataframe.iloc[idx]['filename']\n        img_path = os.path.join(self.root_dir, img_name)\n        \n        # Load image\n        image = cv2.imread(img_path)\n        image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)\n        image = Image.fromarray(image)\n        \n        if self.transform:\n            image = self.transform(image)\n        \n        if self.is_test:\n            return image, img_name\n        else:\n            label = self.dataframe.iloc[idx]['label_encoded']\n            return image, label\n\n# Data augmentation and preprocessing\ndef get_transforms(phase='train'):\n    if phase == 'train':\n        return transforms.Compose([\n            transforms.Resize((224, 224)),\n            transforms.RandomHorizontalFlip(p=0.5),\n            transforms.RandomVerticalFlip(p=0.3),\n            transforms.RandomRotation(degrees=15),\n            transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2, hue=0.1),\n            transforms.RandomAffine(degrees=0, translate=(0.1, 0.1)),\n            transforms.ToTensor(),\n            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n        ])\n    else:\n        return 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        ])\n\n# Model definition using EfficientNet\nclass ButterflyClassifier(nn.Module):\n    def __init__(self, num_classes, model_name='efficientnet_b0'):\n        super(ButterflyClassifier, self).__init__()\n        \n        # Load pre-trained model\n        if model_name == 'efficientnet_b0':\n            self.backbone = models.efficientnet_b0(pretrained=True)\n            self.backbone.classifier = nn.Sequential(\n                nn.Dropout(0.2),\n                nn.Linear(self.backbone.classifier[1].in_features, num_classes)\n            )\n        elif model_name == 'resnet50':\n            self.backbone = models.resnet50(pretrained=True)\n            self.backbone.fc = nn.Linear(self.backbone.fc.in_features, num_classes)\n        elif model_name == 'vit_b_16':\n            self.backbone = models.vit_b_16(pretrained=True)\n            self.backbone.heads = nn.Linear(self.backbone.heads.head.in_features, num_classes)\n        \n    def forward(self, x):\n        return self.backbone(x)\n\n# Training function\ndef train_model(model, train_loader, val_loader, criterion, optimizer, scheduler, num_epochs=25):\n    best_acc = 0.0\n    best_model_wts = model.state_dict()\n    \n    train_losses = []\n    val_losses = []\n    train_accs = []\n    val_accs = []\n    \n    for epoch in range(num_epochs):\n        print(f'Epoch {epoch+1}/{num_epochs}')\n        print('-' * 10)\n        \n        # Training phase\n        model.train()\n        running_loss = 0.0\n        running_corrects = 0\n        \n        for inputs, labels in train_loader:\n            inputs = inputs.to(device)\n            labels = labels.to(device)\n            \n            optimizer.zero_grad()\n            \n            outputs = model(inputs)\n            _, preds = torch.max(outputs, 1)\n            loss = criterion(outputs, labels)\n            \n            loss.backward()\n            optimizer.step()\n            \n            running_loss += loss.item() * inputs.size(0)\n            running_corrects += torch.sum(preds == labels.data)\n        \n        train_loss = running_loss / len(train_loader.dataset)\n        train_acc = running_corrects.double() / len(train_loader.dataset)\n        \n        # Validation phase\n        model.eval()\n        running_loss = 0.0\n        running_corrects = 0\n        \n        with torch.no_grad():\n            for inputs, labels in val_loader:\n                inputs = inputs.to(device)\n                labels = labels.to(device)\n                \n                outputs = model(inputs)\n                _, preds = torch.max(outputs, 1)\n                loss = criterion(outputs, labels)\n                \n                running_loss += loss.item() * inputs.size(0)\n                running_corrects += torch.sum(preds == labels.data)\n        \n        val_loss = running_loss / len(val_loader.dataset)\n        val_acc = running_corrects.double() / len(val_loader.dataset)\n        \n        train_losses.append(train_loss)\n        val_losses.append(val_loss)\n        train_accs.append(train_acc)\n        val_accs.append(val_acc)\n        \n        print(f'Train Loss: {train_loss:.4f} Acc: {train_acc:.4f}')\n        print(f'Val Loss: {val_loss:.4f} Acc: {val_acc:.4f}')\n        \n        # Deep copy the model\n        if val_acc > best_acc:\n            best_acc = val_acc\n            best_model_wts = model.state_dict()\n        \n        scheduler.step()\n        print()\n    \n    print(f'Best val Acc: {best_acc:.4f}')\n    \n    # Load best model weights\n    model.load_state_dict(best_model_wts)\n    return model, train_losses, val_losses, train_accs, val_accs\n\n# Cross-validation training\ndef cross_validation_training(train_df, num_folds=5):\n    skf = StratifiedKFold(n_splits=num_folds, shuffle=True, random_state=42)\n    \n    fold_accuracies = []\n    fold_models = []\n    \n    for fold, (train_idx, val_idx) in enumerate(skf.split(train_df, train_df['label_encoded'])):\n        print(f\"\\nFold {fold + 1}/{num_folds}\")\n        print(\"=\" * 50)\n        \n        train_fold = train_df.iloc[train_idx].reset_index(drop=True)\n        val_fold = train_df.iloc[val_idx].reset_index(drop=True)\n        \n        # Create datasets\n        train_dataset = ButterflyDataset(train_fold, TRAIN_DIR, get_transforms('train'))\n        val_dataset = ButterflyDataset(val_fold, TRAIN_DIR, get_transforms('val'))\n        \n        # Create data loaders\n        train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True, num_workers=4)\n        val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False, num_workers=4)\n        \n        # Initialize model\n        model = ButterflyClassifier(num_classes, 'efficientnet_b0').to(device)\n        \n        # Loss and optimizer\n        criterion = nn.CrossEntropyLoss()\n        optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-4)\n        scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=7, gamma=0.1)\n        \n        # Train model\n        model, _, _, _, _ = train_model(model, train_loader, val_loader, criterion, optimizer, scheduler, num_epochs=15)\n        \n        # Evaluate on validation set\n        model.eval()\n        val_preds = []\n        val_labels = []\n        \n        with torch.no_grad():\n            for inputs, labels in val_loader:\n                inputs = inputs.to(device)\n                outputs = model(inputs)\n                _, preds = torch.max(outputs, 1)\n                val_preds.extend(preds.cpu().numpy())\n                val_labels.extend(labels.numpy())\n        \n        fold_accuracy = accuracy_score(val_labels, val_preds)\n        fold_accuracies.append(fold_accuracy)\n        fold_models.append(model)\n        \n        print(f\"Fold {fold + 1} Accuracy: {fold_accuracy:.4f}\")\n    \n    print(f\"\\nCross-validation mean accuracy: {np.mean(fold_accuracies):.4f} (+/- {np.std(fold_accuracies) * 2:.4f})\")\n    return fold_models, fold_accuracies\n\n# Ensemble prediction\ndef ensemble_predict(models, test_loader, label_encoder):\n    all_predictions = []\n    \n    for model in models:\n        model.eval()\n        predictions = []\n        \n        with torch.no_grad():\n            for inputs, filenames in test_loader:\n                inputs = inputs.to(device)\n                outputs = model(inputs)\n                probabilities = torch.softmax(outputs, dim=1)\n                predictions.append(probabilities.cpu().numpy())\n        \n        all_predictions.append(np.vstack(predictions))\n    \n    # Average predictions across all models\n    ensemble_predictions = np.mean(all_predictions, axis=0)\n    final_predictions = np.argmax(ensemble_predictions, axis=1)\n    \n    # Convert back to original labels\n    final_labels = label_encoder.inverse_transform(final_predictions)\n    \n    return final_labels\n\n# Main execution\nif __name__ == \"__main__\":\n    # Train models using cross-validation\n    print(\"Starting cross-validation training...\")\n    fold_models, fold_accuracies = cross_validation_training(train_df, num_folds=5)\n    \n    # Create test dataset\n    test_dataset = ButterflyDataset(test_df, TEST_DIR, get_transforms('val'), is_test=True)\n    test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False, num_workers=4)\n    \n    # Make ensemble predictions\n    print(\"Making ensemble predictions...\")\n    ensemble_predictions = ensemble_predict(fold_models, test_loader, label_encoder)\n    \n    # Create submission file\n    submission = pd.DataFrame({\n        'filename': test_df['filename'],\n        'label': ensemble_predictions\n    })\n    \n    submission.to_csv('submission.csv', index=False)\n    print(\"Submission file created: submission.csv\")\n    \n    # Display sample predictions\n    print(\"\\nSample predictions:\")\n    print(submission.head(10))\n    \n    # Save the best model\n    best_fold_idx = np.argmax(fold_accuracies)\n    torch.save(fold_models[best_fold_idx].state_dict(), 'best_butterfly_model.pth')\n    print(f\"Best model saved (Fold {best_fold_idx + 1} with accuracy {fold_accuracies[best_fold_idx]:.4f})\")\n\n# Additional utility functions for post-processing and analysis\ndef plot_training_curves(train_losses, val_losses, train_accs, val_accs):\n    \"\"\"Plot training and validation curves\"\"\"\n    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 5))\n    \n    ax1.plot(train_losses, label='Training Loss')\n    ax1.plot(val_losses, label='Validation Loss')\n    ax1.set_title('Training and Validation Loss')\n    ax1.set_xlabel('Epoch')\n    ax1.set_ylabel('Loss')\n    ax1.legend()\n    \n    ax2.plot(train_accs, label='Training Accuracy')\n    ax2.plot(val_accs, label='Validation Accuracy')\n    ax2.set_title('Training and Validation Accuracy')\n    ax2.set_xlabel('Epoch')\n    ax2.set_ylabel('Accuracy')\n    ax2.legend()\n    \n    plt.tight_layout()\n    plt.show()\n\ndef analyze_predictions(model, val_loader, label_encoder):\n    \"\"\"Analyze model predictions and create confusion matrix\"\"\"\n    model.eval()\n    all_preds = []\n    all_labels = []\n    \n    with torch.no_grad():\n        for inputs, labels in val_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    # Convert to original labels\n    pred_labels = label_encoder.inverse_transform(all_preds)\n    true_labels = label_encoder.inverse_transform(all_labels)\n    \n    # Classification report\n    print(\"Classification Report:\")\n    print(classification_report(true_labels, pred_labels))\n    \n    # Confusion matrix\n    plt.figure(figsize=(12, 10))\n    cm = confusion_matrix(true_labels, pred_labels)\n    sns.heatmap(cm, annot=True, fmt='d', cmap='Blues', \n                xticklabels=label_encoder.classes_, \n                yticklabels=label_encoder.classes_)\n    plt.title('Confusion Matrix')\n    plt.xlabel('Predicted')\n    plt.ylabel('Actual')\n    plt.xticks(rotation=45)\n    plt.yticks(rotation=0)\n    plt.tight_layout()\n    plt.show()\n\n# Test Time Augmentation (TTA) for better predictions\ndef tta_predict(model, test_loader, label_encoder, num_tta=5):\n    \"\"\"Test Time Augmentation for more robust predictions\"\"\"\n    model.eval()\n    all_tta_predictions = []\n    \n    tta_transforms = transforms.Compose([\n        transforms.Resize((224, 224)),\n        transforms.RandomHorizontalFlip(p=0.5),\n        transforms.RandomRotation(degrees=10),\n        transforms.ColorJitter(brightness=0.1, contrast=0.1),\n        transforms.ToTensor(),\n        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])\n    ])\n    \n    for tta_idx in range(num_tta):\n        predictions = []\n        \n        with torch.no_grad():\n            for inputs, filenames in test_loader:\n                inputs = inputs.to(device)\n                outputs = model(inputs)\n                probabilities = torch.softmax(outputs, dim=1)\n                predictions.append(probabilities.cpu().numpy())\n        \n        all_tta_predictions.append(np.vstack(predictions))\n    \n    # Average TTA predictions\n    tta_ensemble = np.mean(all_tta_predictions, axis=0)\n    final_predictions = np.argmax(tta_ensemble, axis=1)\n    final_labels = label_encoder.inverse_transform(final_predictions)\n    \n    return final_labels","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-07-07T04:23:52.981744Z","iopub.execute_input":"2025-07-07T04:23:52.982029Z","iopub.status.idle":"2025-07-07T04:49:12.168811Z","shell.execute_reply.started":"2025-07-07T04:23:52.982007Z","shell.execute_reply":"2025-07-07T04:49:12.167897Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}