{"metadata":{"kernelspec":{"display_name":"Python 3 (ipykernel)","language":"python","name":"python3"},"language_info":{"codemirror_mode":{"name":"ipython","version":3},"file_extension":".py","mimetype":"text/x-python","name":"python","nbconvert_exporter":"python","pygments_lexer":"ipython3","version":"3.11.9"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":3175,"databundleVersionId":44352,"sourceType":"competition"}],"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"### Galaxy Zoo - The Galaxy Challenge\n\n#### Author: Ching-Yin NG\n#### Source code and report are available at https://github.com/alvinng4/GalaxyZooChallenge\n#### Best model score: 0.07972\n#### Best score after model averaging: 0.07716","metadata":{}},{"cell_type":"markdown","source":"Q1. Is the object a smooth galaxy, a galaxy with features/disk or a star? 3 responses\\\nQ2. Is it edge-on? 2 responses\\\nQ3. Is there a bar? 2 responses\\\nQ4. Is there a spiral pattern? 2 responses\\\nQ5. How prominent is the central bulge? 4 responses\\\nQ6. Is there anything \"odd\" about the galaxy? 2 responses\\\nQ7. How round is the smooth galaxy? 3 responses\\\nQ8. What is the odd feature? 7 responses\\\nQ9. What shape is the bulge in the edge-on galaxy? 3 responses\\\nQ10. How tightly wound are the spiral arms? 3 responses\\\nQ11. How many spiral arms are there? 6 responses","metadata":{}},{"cell_type":"code","source":"import math\nfrom pathlib import Path\nimport random\nimport timeit\n\nimport matplotlib.pyplot as plt\nimport numpy as np\nfrom PIL import Image\nimport polars\nimport torch\nimport torch.nn as nn\nfrom torch.utils.data import Dataset\nfrom torch.utils.data import DataLoader\n\nSUBCLASS_LEN = [3, 2, 2, 2, 4, 2, 3, 7, 3, 3, 6]\nSUBCLASS_LEN_CUMSUM = np.cumsum(SUBCLASS_LEN)\nCLASS_LEN = len(SUBCLASS_LEN)\n\n\nRANDOM_STATE = 1\ntorch.manual_seed(RANDOM_STATE)\n\nFILE_PREFIX = \"test\"\nMODEL_DIR = Path(f\"models/{FILE_PREFIX}\")\nIMAGE_DIR = Path(\"images\")\n\nMODEL_DIR.mkdir(parents=True, exist_ok=True)\nIMAGE_DIR.mkdir(parents=True, exist_ok=True)\n\nprint(torch.cuda.is_available())\nif torch.cuda.is_available():\n    device = torch.device(\"cuda:0\")\nelse:\n    device = torch.device(\"cpu\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Import dataset","metadata":{}},{"cell_type":"code","source":"data_folder = Path(\"../data\")\n\nimg_path = data_folder / \"images_training_rev1\"\nfile_list = sorted([str(path) for path in img_path.glob('*.jpg')])\nprint(f\"Number of data: {len(file_list)}\")\n\ntrain_sol_path = data_folder / \"training_solutions_rev1\" / \"training_solutions_rev1.csv\"\ntrain_sol = polars.read_csv(train_sol_path)\nprint(\"Some labels: \")\nprint(train_sol.head())\n\nsoft_labels = train_sol[:, 1:].to_numpy()\n\none_hot_labels = np.argmax(soft_labels[:, :SUBCLASS_LEN_CUMSUM[0]], axis=1)    # Class one\nfor i in range(1, CLASS_LEN):\n    one_hot_labels = np.vstack(\n        (one_hot_labels, np.argmax(soft_labels[:, SUBCLASS_LEN_CUMSUM[i - 1]:SUBCLASS_LEN_CUMSUM[i]], axis=1))\n    )\none_hot_labels = one_hot_labels.T\nprint(soft_labels)\nprint(soft_labels.shape)\nprint(one_hot_labels)\nprint(one_hot_labels.shape)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from collections import Counter\n\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.utils import resample\nfrom torchvision import transforms\n\nclass ImageDataset(Dataset):\n    def __init__(self, file_list, labels, transform=None):\n        self.file_list = file_list\n        self.labels = labels\n        self.transform = transform\n\n    def __getitem__(self, index):\n        file = self.file_list[index]\n        label = self.labels[index]\n\n        image = Image.open(file)\n\n        if self.transform is not None:\n            image = self.transform(image)\n\n        return image, label\n    \n    def __len__(self):\n        return len(self.labels)\n        \ntrain_file_list, temp_file_list, train_soft_labels, temp_soft_labels =\\\n     train_test_split(file_list, soft_labels, test_size=0.1,\n                      random_state=RANDOM_STATE, stratify=one_hot_labels[:, 0])\n\ntemp_one_hot_labels = np.argmax(temp_soft_labels[:, :3], axis=1)\n\nvalid_file_list, test_file_list, valid_soft_labels, test_soft_labels =\\\n     train_test_split(temp_file_list, temp_soft_labels, test_size=0.7,\n                      random_state=RANDOM_STATE, stratify=temp_one_hot_labels)\n\ntransform_train = transforms.Compose([\n    transforms.RandomRotation(360),\n    transforms.CenterCrop([256, 256]),\n    transforms.Resize([128, 128]),\n    transforms.RandomHorizontalFlip(),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=(0.5, 0.5, 0.5), std=(0.5, 0.5, 0.5))\n])\n\ntransform = transforms.Compose([\n    transforms.CenterCrop([256, 256]),\n    transforms.Resize([128, 128]),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=(0.5, 0.5, 0.5), std=(0.5, 0.5, 0.5))\n])\n\ntrain_dataset = ImageDataset(train_file_list, train_soft_labels, transform_train)\nvalid_dataset = ImageDataset(valid_file_list, valid_soft_labels, transform)\ntest_dataset = ImageDataset(test_file_list, test_soft_labels, transform)\n\nprint(f\"Length of training files: {len(train_file_list)}\")\nprint(f\"Length of validation files: {len(valid_file_list)}\")\nprint(f\"Length of testing files: {len(test_file_list)}\")\n\ntrain_one_hot_labels_class_one = np.argmax(train_soft_labels[:, :3], axis=1)\nvalid_one_hot_labels_class_one = np.argmax(valid_soft_labels[:, :3], axis=1)\ntest_one_hot_labels_class_one = np.argmax(test_soft_labels[:, :3], axis=1)\n\nprint()\nprint(\"Class 1 distribution: \")\nprint(f\"Training dataset: {Counter(train_one_hot_labels_class_one)}\")\nprint(f\"Validation dataset: {Counter(valid_one_hot_labels_class_one)}\")\nprint(f\"Testing dataset: {Counter(test_one_hot_labels_class_one)}\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"data_loader = DataLoader(train_dataset, batch_size=100, shuffle=False)\n\nimg_class_zero = []\nimg_class_one = []\nimg_class_two = []\nimg_per_class = 5\n\niter_data_loader = iter(data_loader)\nwhile True:\n    img_batch, label_batch = next(iter_data_loader)\n    for i in range(100):\n        img = img_batch[i]\n        label = label_batch[i]\n        one_hot_label = np.argmax(label[:3])\n\n        if (len(img_class_zero) < img_per_class) and (one_hot_label == 0):\n            img_class_zero.append(img)\n        elif (len(img_class_one) < img_per_class) and (one_hot_label == 1):\n            img_class_one.append(img)\n        elif (len(img_class_two) < img_per_class) and (one_hot_label == 2):\n            img_class_two.append(img)\n            \n    if (\n        (len(img_class_zero) >= img_per_class)\n        and (len(img_class_one) >= img_per_class)\n        and (len(img_class_two) >= img_per_class)\n    ):\n        break\n\nfig1 = plt.figure(figsize=(9, 15))\nfor i in range(img_per_class):\n    img = img_class_zero[i].cpu()\n    ax = fig1.add_subplot(img_per_class, 3, (i * 3) + 1)\n    ax.set_xticks([])\n    ax.set_yticks([])\n    ax.set_title(f'Class 1.1:', size=15)\n    ax.imshow(img.permute(1, 2, 0))\n\n    img = img_class_one[i].cpu()\n    ax = fig1.add_subplot(img_per_class, 3, (i * 3) + 2)\n    ax.set_xticks([])\n    ax.set_yticks([])\n    ax.set_title(f'Class 1.2:', size=15)\n    ax.imshow(img.permute(1, 2, 0))\n\n    img = img_class_two[i].cpu()\n    ax = fig1.add_subplot(img_per_class, 3, (i * 3) + 3)\n    ax.set_xticks([])\n    ax.set_yticks([])\n    ax.set_title(f'Class 1.3:', size=15)\n    ax.imshow(img.permute(1, 2, 0))\n\nplt.show()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Creating a CNN-CBAM model","metadata":{}},{"cell_type":"code","source":"class channel_attention(nn.Module):\n    def __init__(self, in_channels, reduction_ratio):\n        super().__init__()\n        self.max_pool = nn.AdaptiveMaxPool2d(1)\n        self.avg_pool = nn.AdaptiveAvgPool2d(1)\n        self.flatten = nn.Flatten()\n        hidden_layer_size = int(in_channels // reduction_ratio)\n\n        self.mlp = nn.Sequential()\n        self.mlp.add_module(\"fc1\", nn.Linear(in_channels, hidden_layer_size))\n        self.mlp.add_module(\"relu\", nn.ReLU())\n        self.mlp.add_module(\"fc2\", nn.Linear(hidden_layer_size, in_channels))\n        self.sigmoid = nn.Sigmoid()\n\n    def forward(self, x):\n        max_pooled_map = self.max_pool(x)\n        avg_pooled_map = self.avg_pool(x)\n\n        max_att_map = self.mlp(self.flatten(max_pooled_map))\n        avg_att_map = self.mlp(self.flatten(avg_pooled_map))\n\n        cha_att_map = self.sigmoid(max_att_map + avg_att_map).unsqueeze(2).unsqueeze(3)\n\n        return cha_att_map\n\nclass spatial_attention(nn.Module):\n    def __init__(self, in_channels, kernel_size):\n        super().__init__()\n        self.conv = nn.Conv2d(in_channels=2, out_channels=1, kernel_size=kernel_size, padding=((kernel_size - 1) // 2))\n        self.sigmoid = nn.Sigmoid()\n\n    def forward(self, x):\n        max_pooled_map = torch.amax(x, dim=1, keepdim=True)\n        avg_pooled_map = torch.mean(x, dim=1, keepdim=True)\n\n        concat_pooled_map = torch.cat((max_pooled_map, avg_pooled_map), dim=1)\n        conv_map = self.conv(concat_pooled_map)\n        spa_att_map = self.sigmoid(conv_map)\n\n        return spa_att_map\n\nclass CBAM(nn.Module):\n    def __init__(self, in_channels, cha_att_reduction_ratio, spa_att_kernel_size):\n        super().__init__()\n        self.channel_attention_module = channel_attention(\n            in_channels, cha_att_reduction_ratio\n        )\n        self.spatial_attention_module = spatial_attention(\n            in_channels, spa_att_kernel_size\n        )\n\n    def forward(self, x):\n        cha_att_map = self.channel_attention_module(x)\n        spa_att_map = self.spatial_attention_module(x * cha_att_map)\n\n        return x + (x * cha_att_map * spa_att_map)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model = nn.Sequential()\n\nmodel.add_module('conv1', nn.Conv2d(in_channels=3, out_channels=32, kernel_size=17, padding=8))\nmodel.add_module('bn1', nn.BatchNorm2d(num_features=32))\nmodel.add_module('relu1', nn.ReLU())       \nmodel.add_module('cbam1', CBAM(in_channels=32, cha_att_reduction_ratio=1, spa_att_kernel_size=7))\nmodel.add_module('pool1', nn.MaxPool2d(kernel_size=2))  \n# model.add_module('dropout1', nn.Dropout2d(p=0.15))\n\nmodel.add_module('conv2', nn.Conv2d(in_channels=32, out_channels=64, kernel_size=13, padding=6))\nmodel.add_module('bn2', nn.BatchNorm2d(num_features=64))\nmodel.add_module('relu2', nn.ReLU())        \nmodel.add_module('cbam2', CBAM(in_channels=64, cha_att_reduction_ratio=1, spa_att_kernel_size=7))\nmodel.add_module('pool2', nn.MaxPool2d(kernel_size=2))\n# model.add_module('dropout2', nn.Dropout2d(p=0.15))\n\nmodel.add_module('conv3', nn.Conv2d(in_channels=64, out_channels=128, kernel_size=9, padding=4))\nmodel.add_module('bn3', nn.BatchNorm2d(num_features=128))\nmodel.add_module('relu3', nn.ReLU())        \nmodel.add_module('cbam3', CBAM(in_channels=128, cha_att_reduction_ratio=1, spa_att_kernel_size=7))\nmodel.add_module('pool3', nn.MaxPool2d(kernel_size=2))\n# model.add_module('dropout3', nn.Dropout2d(p=0.15))\n\nmodel.add_module('conv4', nn.Conv2d(in_channels=128, out_channels=256, kernel_size=7, padding=3))\nmodel.add_module('bn4', nn.BatchNorm2d(num_features=256))\nmodel.add_module('relu4', nn.ReLU())\nmodel.add_module('cbam4', CBAM(in_channels=256, cha_att_reduction_ratio=1, spa_att_kernel_size=7))","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"x = torch.ones((4, 3, 128, 128))\nprint(model(x).shape)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.add_module('pool4', nn.AvgPool2d(kernel_size=2)) \nmodel.add_module('flatten', nn.Flatten())\n\nx = torch.ones((4, 3, 128, 128))\nprint(model(x).shape)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.add_module('fc1', nn.Linear(16384, 1024)) \n# model.add_module('bn5', nn.BatchNorm1d(num_features=1024))\n# model.add_module('dropout4', nn.Dropout(p=0.2))\n\nx = torch.ones((4, 3, 128, 128))\nprint(model(x).shape)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model.add_module('fc2', nn.Linear(1024, SUBCLASS_LEN_CUMSUM[-1]) )\nmodel.add_module('sigmoid', nn.Sigmoid())\n\nx = torch.ones((4, 3, 128, 128))\nprint(model(x).shape)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(model)\n\ntotal_params = sum(p.numel() for p in model.parameters())\nprint(f\"Number of parameters: {total_params}\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"model = model.to(device)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Training","metadata":{}},{"cell_type":"code","source":"hist = [[], [], [], []]\nlr_hist = []","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from sklearn.metrics import mean_squared_error\n\ndef train(model, num_epochs, lr_scheduler, optimizer, train_dl, valid_dl, batch_size, hist, lr_hist, save_every_n_epochs):\n    loss_fn = nn.BCELoss()\n    previous_epochs = len(hist[0])\n    \n    for epoch in range(num_epochs):\n        start = timeit.default_timer()\n        model.train()\n        i = 0\n        loss_hist_train = 0\n        rmse_hist_train = 0\n        for x_batch, y_batch in train_dl:\n            i += 1\n            if i % 5 == 0:\n                print(f\"Progress: epoch: {previous_epochs + epoch + 1}/{previous_epochs + num_epochs}, batch: {i}/{len(train_dl)}\", end=\"\\r\")\n            x_batch = x_batch.to(device)\n            y_batch = y_batch.to(device)\n            pred = model(x_batch)\n            loss = loss_fn(pred.float(), y_batch.float())\n            loss.backward()\n            optimizer.step()\n            optimizer.zero_grad()\n            \n            loss_hist_train += loss.item() * y_batch.size(0)\n            rmse_hist_train += mean_squared_error(y_true=y_batch.cpu(), y_pred=pred.detach().cpu())\n\n        loss_hist_train /= len(train_dl.dataset)\n        rmse_hist_train /= (len(train_dl.dataset) / batch_size)\n        rmse_hist_train = math.sqrt(rmse_hist_train)\n        \n        model.eval()\n        loss_hist_valid = 0\n        rmse_hist_valid = 0\n        with torch.no_grad():\n            for x_batch, y_batch in valid_dl:\n                x_batch = x_batch.to(device)\n                y_batch = y_batch.to(device)\n                pred = model(x_batch)\n                loss = loss_fn(pred.float(), y_batch.float())\n                \n                loss_hist_valid += loss.item() * y_batch.size(0)\n                rmse_hist_valid += mean_squared_error(y_true=y_batch.cpu(), y_pred=pred.detach().cpu())\n\n        loss_hist_valid /= len(valid_dl.dataset)\n        rmse_hist_valid /= (len(valid_dl.dataset) / batch_size)\n        rmse_hist_valid = math.sqrt(rmse_hist_valid)\n\n        end = timeit.default_timer()\n\n        print_str = (\n            f\"Epoch: {previous_epochs + epoch + 1} \" +\n            f\"train rmse: {rmse_hist_train:.4f} \" +\n            f\"train loss: {loss_hist_train:.4f} \" +\n            f\"val rmse: {rmse_hist_valid:.4f} \" +\n            f\"val loss: {loss_hist_valid:.4f} \" +\n            f\"lr: {lr_scheduler.get_last_lr()[0]} \" +\n            f\"Time taken: {end - start:.3f}s\"\n        )\n        print(print_str)\n\n        if (previous_epochs + epoch + 1) % save_every_n_epochs == 0:\n            torch.save(model, MODEL_DIR / (FILE_PREFIX + f\"_epoch{previous_epochs + epoch + 1}.pth\"))\n\n        results = [loss_hist_train, loss_hist_valid, rmse_hist_train, rmse_hist_valid]\n        for i in range(len(hist)):\n            hist[i].append(results[i])\n        lr_hist.append(lr_scheduler.get_last_lr()[0])\n\n        lr_scheduler.step()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Ensure that the minority class appears in each (n) batch\nfrom operator import itemgetter\n\nn = 10\n\ncounter = Counter(train_one_hot_labels_class_one)\nmin_key, min_count = min(counter.items(), key=itemgetter(1))\nbatch_size = round(sum(counter.values()) / (min_count * n))\nprint(batch_size)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"num_workers = 8    # num_workers must set to zero if the images are already transferred to the GPU\n\nif num_workers == 0:\n    train_dl = DataLoader(train_dataset, batch_size, num_workers=0, shuffle=True)\n    valid_dl = DataLoader(valid_dataset, batch_size, num_workers=0, shuffle=False)\nelse:\n    train_dl = DataLoader(train_dataset, batch_size, num_workers=num_workers, persistent_workers=True, pin_memory=True, shuffle=True)\n    valid_dl = DataLoader(valid_dataset, batch_size, num_workers=num_workers, persistent_workers=True, pin_memory=True, shuffle=False)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from torch.optim import Adam\nfrom torch.optim.lr_scheduler import CosineAnnealingWarmRestarts\n\nlearning_rate = 0.0001\n\noptimizer = Adam(model.parameters(), lr=learning_rate)\nlr_scheduler = CosineAnnealingWarmRestarts(optimizer, T_0=60, T_mult=3, eta_min=1e-8)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"num_epochs = 240\nsave_every_n_epochs = 10\n\ntrain(model, num_epochs, lr_scheduler, optimizer,\\\n      train_dl, valid_dl, batch_size, hist, lr_hist, save_every_n_epochs)","metadata":{"scrolled":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"x_arr = np.arange(len(hist[0])) + 1\n\nfig = plt.figure(figsize=(12, 4))\nax = fig.add_subplot(1, 3, 1)\nax.plot(x_arr, hist[0], '-o', label='Train loss')\nax.plot(x_arr, hist[1], '--<', label='Validation loss')\nax.legend(fontsize=15)\nax.set_xlabel('Epoch', size=15)\nax.set_ylabel('Loss', size=15)\n\nax = fig.add_subplot(1, 3, 2)\nax.plot(x_arr, hist[2], '-o', label='Train RMSE')\nax.plot(x_arr, hist[3], '--<', label='Validation RMSE')\nax.legend(fontsize=15)\nax.set_xlabel('Epoch', size=15)\nax.set_ylabel('RMSE', size=15)\n\nax = fig.add_subplot(1, 3, 3)\nax.plot(x_arr, lr_hist, '-o', label='lr rate')\nax.legend(fontsize=15)\nax.set_xlabel('Epoch', size=15)\nax.set_ylabel('learning rate', size=15)\n\n\nplt.tight_layout()\nplt.savefig(IMAGE_DIR / f\"{FILE_PREFIX}_{learning_rate}.png\")\nplt.show()","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from sklearn.metrics import confusion_matrix, classification_report, root_mean_squared_error\n\ntest_dl = DataLoader(test_dataset, batch_size, num_workers=0, pin_memory=True, shuffle=False)\n\nprint(\"Getting test results from model...\")\nmodel.eval()\nwith torch.no_grad():\n    all_preds_soft_labels = np.zeros((len(test_dataset), SUBCLASS_LEN_CUMSUM[-1]))\n    all_soft_labels = np.zeros((len(test_dataset), SUBCLASS_LEN_CUMSUM[-1]))\n    i = 0\n    for x_batch, y_batch in test_dl:\n        print(f\"Progress: {i} / {len(test_dataset)}\", end=\"\\r\")\n        x_batch = x_batch.to(device) \n        y_batch = y_batch.to(device) \n\n        all_preds_soft_labels[i: (i + batch_size)] = model(x_batch).cpu()\n        all_soft_labels[i: (i + batch_size)] = y_batch.cpu()\n\n        i += batch_size\n\nall_preds_one_hot_labels_class_one = np.argmax(all_preds_soft_labels[:, :3], axis=1)\nall_one_hot_labels_class_one = np.argmax(all_soft_labels[:, :3], axis=1)\n\nrmse = root_mean_squared_error(all_soft_labels, all_preds_soft_labels)\nrmse_per_column = np.zeros(all_soft_labels.shape[1])\nfor i in range(all_soft_labels.shape[1]):\n    rmse_per_column[i] = root_mean_squared_error(all_soft_labels[:, i], all_preds_soft_labels[:, i])\n\nprint()\nprint(f\"RMSE: {rmse}\")\nprint(f\"RMSE per column: {rmse_per_column}\")\nprint()\nprint(\"Class one:\")\nprint(f\"Confusion matrix: \\n{confusion_matrix(all_one_hot_labels_class_one, all_preds_one_hot_labels_class_one, labels=[0, 1, 2])}\")\nprint(classification_report(all_one_hot_labels_class_one, all_preds_one_hot_labels_class_one, labels=[0, 1, 2]))","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Test submission","metadata":{}},{"cell_type":"code","source":"final_test_img_path = data_folder / \"images_test_rev1\"\nfinal_test_file_list = sorted([str(path) for path in final_test_img_path.glob('*.jpg')])\nprint(f\"Number of test files: {len(final_test_file_list)}\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class FinalTestImageDataset(Dataset):\n    def __init__(self, file_list, transform=None):\n        self.file_list = file_list\n        self.transform = transform\n\n    def __getitem__(self, index):\n        file = self.file_list[index]\n        image = Image.open(file)\n\n        if self.transform is not None:\n            image = self.transform(image)\n\n        gxy_id = Path(file).stem\n        \n        return image, gxy_id\n    \n    def __len__(self):\n        return len(self.file_list)\n\nfinal_test_dataset = FinalTestImageDataset(final_test_file_list, transform)\nfinal_test_dl = DataLoader(final_test_dataset, num_workers=8, persistent_workers=True, batch_size=1000, shuffle=False)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"import csv\n\nwith open(\"submission.csv\", \"w\", newline=\"\") as submission_file:\n    writer = csv.writer(submission_file)\n    writer.writerow([\n        \"GalaxyID\", \"Class1.1\", \"Class1.2\", \"Class1.3\", \"Class2.1\", \"Class2.2\", \n        \"Class3.1\", \"Class3.2\", \"Class4.1\", \"Class4.2\", \"Class5.1\", \"Class5.2\", \n        \"Class5.3\", \"Class5.4\", \"Class6.1\", \"Class6.2\", \"Class7.1\", \"Class7.2\", \n        \"Class7.3\", \"Class8.1\", \"Class8.2\", \"Class8.3\", \"Class8.4\", \"Class8.5\", \n        \"Class8.6\", \"Class8.7\", \"Class9.1\", \"Class9.2\", \"Class9.3\", \"Class10.1\", \n        \"Class10.2\", \"Class10.3\", \"Class11.1\", \"Class11.2\", \"Class11.3\", \"Class11.4\", \n        \"Class11.5\", \"Class11.6\"\n    ])\n\n    print(\"Getting test results from model...\")\n    model.eval()\n    all_preds_soft_labels = np.zeros((len(final_test_dataset), SUBCLASS_LEN_CUMSUM[-1]))\n    gxy_ids = np.zeros((len(final_test_dataset),))\n    i = 0\n    with torch.no_grad():\n        for img_batch, gxy_id_batch in final_test_dl:\n            print(f\"Progress: {i + 1} / {len(all_preds_soft_labels)}\", end=\"\\r\")\n            img_batch = img_batch.to(device) \n            pred = model(img_batch)\n            pred_soft_labels = pred.cpu().numpy()\n\n            if len(pred_soft_labels) == 1000:\n                all_preds_soft_labels[i: (i + 1000)] = pred_soft_labels\n                gxy_ids[i: (i + 1000)] = gxy_id_batch\n                i += 1000\n            else:\n                all_preds_soft_labels[i:] = pred_soft_labels\n                gxy_ids[i: ] = gxy_id_batch\n                i += len(pred_soft_labels)\n                break\n\n    print()\n    print(\"Writing test results to submission file...\")\n    for i in range(len(all_preds_soft_labels)):\n        print(f\"Progress: {i + 1} / {len(all_preds_soft_labels)}\", end=\"\\r\")\n        output_list = [int(gxy_ids[i])] + all_preds_soft_labels[i].tolist()\n        writer.writerow(output_list)\n\n    print()\n    print(\"Done!\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pytorchsummary import summary\n\nsummary(model=model, input_size=(3, 128, 128))\n\nfor name, param in model.named_parameters():\n    print(f\"Parameter: {name}, Mean: {param.data.mean().item()}\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from pytorch_grad_cam import AblationCAM\nfrom pytorch_grad_cam.utils.image import show_cam_on_image\n\ntarget_layers = [model.conv1, model.cbam1]\nprint(target_layers)\n\ncam = AblationCAM(model=model, target_layers=target_layers)\ngrayscale_cam = cam(input_tensor=transform(Image.open(\"../data/images_test_rev1/100037.jpg\")).unsqueeze(0))\n\nresize = transforms.Resize((128, 128))\nimg_array = resize(Image.open(\"../data/images_test_rev1/100037.jpg\"))\nimg_array = np.array(img_array)\nimg_array = (img_array - np.min(img_array)) / (np.max(img_array) - np.min(img_array))\nplt.imshow(img_array)\nplt.show()\nvisualization = show_cam_on_image(img_array, grayscale_cam[0], use_rgb=True)\nplt.figure(figsize=(4, 4))\nplt.imshow(visualization)\nplt.title(\"First layer\")\n# plt.savefig(\"images/ablationcam1\")\nplt.show()","metadata":{},"execution_count":null,"outputs":[]}]}