{"metadata":{"kernelspec":{"display_name":"ai-py310","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.10.13"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceId":29193,"databundleVersionId":2318453,"sourceType":"competition"}],"dockerImageVersionId":30627,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"## Dependencies","metadata":{}},{"cell_type":"code","source":"%matplotlib inline\nimport pandas as pd\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom IPython import display\nimport torch\nfrom torchvision import transforms\nimport torchvision.models as models\nfrom torch.utils.data import Dataset, DataLoader\nimport torch.nn as nn\nimport torch.nn.functional as F\nfrom PIL import Image\nfrom tqdm import tqdm","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Dataset processing","metadata":{}},{"cell_type":"code","source":"INPUT_PATH = '/kaggle/input/classify-leaves/'\nOUTPUT_PATH = '/kaggle/working/'","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# train data\n\ntrain_data = pd.read_csv(INPUT_PATH + 'train.csv')\n\ntrain_img_paths = train_data.iloc[:, 0].values\ntrain_img_labels = train_data.iloc[:, 1].values\nnum_labels = len(np.unique(train_img_labels))\nprint(\"Total label:\", num_labels)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Label map","metadata":{}},{"cell_type":"code","source":"# Label distribution\ndef get_class_map(labels):\n    # sort by frequency\n    unique_labels, counts = np.unique(labels, return_counts=True)\n    # sorted_index = np.argsort(counts)\n    # unique_labels = unique_labels[sorted_index]\n\n    class2num_map: dict[str, int] = dict(zip(unique_labels, range(len(unique_labels))))\n    num2class_map: dict[int, str] = dict(zip(range(len(unique_labels)), unique_labels))\n\n    return class2num_map, num2class_map\n\nclass2num, num2class = get_class_map(train_img_labels)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Test input shape","metadata":{}},{"cell_type":"code","source":"def extract_tensor(index):\n    path = train_data.iloc[index, 0]\n    return torch.tensor(plt.imread(INPUT_PATH + path), dtype=torch.float32).permute(2, 0, 1).unsqueeze(0)\n\nprint(\"Input shape:\", extract_tensor(0).shape)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Define dataset","metadata":{}},{"cell_type":"code","source":"class LeafDataset(Dataset):\n    def __init__(self, csv_path, resize_height, resize_width, valid_ratio=0.2, mode=\"train\"):\n        self.csv_path = csv_path\n        self.resize_height = resize_height\n        self.resize_width = resize_width\n\n        self.valid_ratio = valid_ratio\n        self.mode = mode\n\n        self.data_info = pd.read_csv(self.csv_path)\n        # self.img_paths = self.data_info.iloc[:, 0].values\n\n        self.data_len = len(self.data_info.index)\n        self.train_len = int(self.data_len * (1 - self.valid_ratio))\n\n        def split_data(data: pd.DataFrame, train_len):\n            unique_labels, counts = np.unique(data.iloc[:, 1].values, return_counts=True)\n            # train data should include all classes, if one label is less than 2, then it should be in train data\n            valid_labels_pool = unique_labels[counts > 1]\n            valid_data_pool = data[data.iloc[:, 1].isin(valid_labels_pool)]\n            valid_index_pool = valid_data_pool.index\n            valid_index: list[int] = np.random.choice(valid_index_pool, size=train_len)\n            train_index: list[int] = np.setdiff1d(data.index, valid_index)\n            return data.iloc[train_index], data.iloc[valid_index]\n\n        if self.mode == \"train\":\n            train_data_info, _ = split_data(self.data_info, self.train_len)\n\n            self.img_paths = train_data_info.iloc[:, 0].values\n            self.img_labels = train_data_info.iloc[:, 1].values\n        elif self.mode == \"valid\":\n            _, valid_data_info = split_data(self.data_info, self.train_len)\n\n            self.img_paths = valid_data_info.iloc[:, 0].values\n            self.img_labels = valid_data_info.iloc[:, 1].values\n\n        elif self.mode == \"test\":\n            self.img_paths = self.data_info.iloc[:, 0].values\n            self.img_labels = [\"None\" for _ in range(self.data_len)]  # no label for initialization\n        else:\n            raise ValueError(\"Unknown mode\")\n\n        self.real_len = len(self.img_paths)  # real length of dataset\n\n    def __len__(self):\n        return self.real_len\n\n    def __getitem__(self, idx) -> tuple[torch.Tensor, int | str]:\n        img_path = INPUT_PATH + self.img_paths[idx]\n        img_label = self.img_labels[idx]\n        # ori_img_tensor = torch.tensor(plt.imread(img_path), dtype=torch.float32).permute(2, 0, 1).unsqueeze()\n\n        if self.mode == \"train\" or self.mode == \"valid\":\n            transform = transforms.Compose([\n                transforms.RandomHorizontalFlip(0.5),\n                transforms.RandomVerticalFlip(0.5),\n                transforms.Resize((self.resize_height, self.resize_width)),\n                transforms.ToTensor(),\n                transforms.Normalize(mean=[0.485, 0.456, 0.406],\n                                    std=[0.299, 0.224, 0.225])\n            ])\n            img_tensor: torch.Tensor = transform(Image.open(img_path))\n\n            label_int = class2num[img_label]\n            return img_tensor, label_int\n\n        elif self.mode == \"test\":\n            transform = transforms.Compose([\n                transforms.Resize((self.resize_height, self.resize_width)),\n                transforms.ToTensor(),\n                transforms.Normalize(mean=[0.485, 0.456, 0.406],\n                                    std=[0.299, 0.224, 0.225])\n            ])\n            img_tensor: torch.Tensor = transform(Image.open(img_path))\n\n            if type(img_label) == str: # return label as string\n                return img_tensor, img_label\n            elif type(img_label) == int:\n                return img_tensor, num2class[img_label]\n            else:\n                return img_tensor, img_label\n\n        else:\n            raise ValueError(\"Unknown mode\")\n\n\ntrain_set = LeafDataset(csv_path=INPUT_PATH+\"train.csv\", resize_height=224, resize_width=224, mode=\"train\")\nvalid_set = LeafDataset(csv_path=INPUT_PATH+\"train.csv\", resize_height=224, resize_width=224, mode=\"valid\")\ntest_set = LeafDataset(csv_path=INPUT_PATH+\"test.csv\", resize_height=224, resize_width=224, mode=\"test\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Model","metadata":{}},{"cell_type":"markdown","source":"### Define model (ResNet)","metadata":{}},{"cell_type":"code","source":"IN_CHANNELS = 3\nNUM_LABELS = 176\n\ndef get_model_custom(): # ResNet18\n    # one residual block\n    class Residual(nn.Module):\n        def __init__(self, input_channels, num_channels, strides=1, use_1x1conv=False):\n            super().__init__()\n\n            self.conv1 = nn.Conv2d(input_channels, num_channels,\n                                kernel_size=3, padding=1, stride=strides)\n            self.conv2 = nn.Conv2d(num_channels, num_channels,\n                                    kernel_size=3, padding=1)\n\n            if use_1x1conv:\n                # 1x1 conv: new_X = Y + conv(X)\n                self.res_conv = nn.Conv2d(input_channels, num_channels,\n                                        kernel_size=1, stride=strides)\n\n            self.bn1 = nn.BatchNorm2d(num_channels)\n            self.bn2 = nn.BatchNorm2d(num_channels)\n\n        def forward(self, X):\n            Y = F.relu(self.bn1(self.conv1(X)))\n            Y = self.bn2(self.conv2(Y))\n            if hasattr(self, 'res_conv'):\n                X = self.res_conv(X) # 1x1 conv\n            Y += X\n            return F.relu(Y)\n\n    # ResNet block module\n    def resnet_block(input_channels, num_channels, num_residuals, first_block=False):\n        blk = []\n        for i in range(num_residuals):\n            if i == 0 and not first_block:\n                blk.append(Residual(input_channels, num_channels, strides=2, use_1x1conv=True))\n            else:\n                blk.append(Residual(num_channels, num_channels))\n        return blk\n\n    # ResNet (18)\n    b1 = nn.Sequential(nn.Conv2d(IN_CHANNELS, 64, kernel_size=7, stride=2, padding=3),\n                    nn.BatchNorm2d(64), nn.ReLU(),\n                    nn.MaxPool2d(kernel_size=3, stride=2, padding=1))\n\n    b2 = nn.Sequential(*resnet_block(64, 64, 2, first_block=True), nn.Dropout())\n    b3 = nn.Sequential(*resnet_block(64, 128, 2), nn.Dropout())\n    b4 = nn.Sequential(*resnet_block(128, 256, 2), nn.Dropout())\n    b5 = nn.Sequential(*resnet_block(256, 512, 2), nn.Dropout())\n\n    net = nn.Sequential(b1, b2, b3, b4, b5,\n                        nn.AdaptiveAvgPool2d((1,1)),\n                        nn.Flatten(),\n                        nn.Linear(512, NUM_LABELS))\n\n    return net\n\ndef get_model_ResNet18():\n    net = models.resnet18(pretrained=True)\n    net.fc = nn.Sequential(nn.Linear(net.fc.in_features, NUM_LABELS))\n    return net\n\ndef get_model_ResNet50():\n    net = models.resnext50_32x4d(pretrained=True) # resnext50_32x4d\n    net.fc = nn.Sequential(nn.Linear(net.fc.in_features, NUM_LABELS))\n    return net\n\ndef get_model(name):\n    if name == \"ResNet18\":\n        return get_model_ResNet18()\n    elif name == \"ResNet50\":\n        return get_model_ResNet50()\n    elif name == \"custom\":\n        return get_model_custom()\n    else:\n        raise ValueError(\"Unknown model name\")","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Train","metadata":{}},{"cell_type":"markdown","source":"### Hyperparameters","metadata":{}},{"cell_type":"code","source":"BATCH_SIZE = 16\nNUM_EPOCHS = 50\nLEARNING_RATE = 0.0001\nWEIGHT_DECAY = 0.008","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"print(\"Batch size:\", BATCH_SIZE)\nprint(\"Number of epochs:\", NUM_EPOCHS)\nprint(\"Learning rate:\", LEARNING_RATE)\nprint(\"Weight decay:\", WEIGHT_DECAY)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Training","metadata":{}},{"cell_type":"code","source":"# Get device if CUDA is available\ndef get_device(device='auto'):\n    if not torch.cuda.is_available():\n        device = torch.device('cpu')\n    elif device != 'cpu' and device != 'cuda':\n        device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n    else:\n        device = torch.device(device)\n    return device\n\n\n# Randomly initialize weights\ndef init_weights(m):\n    if type(m) == nn.Linear or type(m) == nn.Conv2d:\n        nn.init.xavier_uniform_(m.weight)  # initialize weights\n\n\ndef train(model, train_set, valid_set, batch_size, num_epochs, lr, weight_decay, device='auto'):\n    device = get_device(device)\n\n    print(\"Training via %s...\" % device.type.upper())\n\n    train_iter = DataLoader(train_set, batch_size=batch_size, num_workers=5, shuffle=False)\n    valid_iter = DataLoader(valid_set, batch_size=batch_size, num_workers=5, shuffle=False)\n\n    model = model.to(device)\n    model.apply(init_weights)  # initialize weights\n\n    loss = nn.CrossEntropyLoss()  # loss function\n    optimizer = torch.optim.Adam(model.parameters(), lr=lr, weight_decay=weight_decay)  # optimizer\n\n    # for paint\n    epoch_list = []\n    y_train_loss_list = []\n    y_valid_loss_list = []\n\n    # train\n    for epoch in range(num_epochs):\n        # train_loss_list: loss of all train batches\n        # train_acc_list: accuracy of all train batches\n        train_loss_list, train_acc_list = [], []\n        # n: total number of train samples\n        n = 0\n        for batch in tqdm(train_iter):\n            X, y = batch\n            X = X.to(device)\n            y = y.to(device)\n            y_hat = model(X)\n\n            l = loss(y_hat, y)\n\n            optimizer.zero_grad()\n            l.backward()\n            optimizer.step()\n\n            train_loss = l.cpu().item()\n            # take max as the predicted label\n            train_acc = (y_hat.argmax(dim=1) == y).sum().cpu().item()\n            train_loss_list.append(train_loss)\n            train_acc_list.append(train_acc)\n\n            n += y.shape[0]\n\n        model.eval()\n\n        # valid_loss_list: loss of all validation batches\n        # valid_acc_list: accuracy of all validation batches\n        valid_acc_list, valid_loss_list = [], []\n        # valid_n: total number of validation samples\n        valid_n = 0\n        for batch in tqdm(valid_iter):\n            X, y = batch\n            X = X.to(device)\n            y = y.to(device)\n            y_hat = model(X)\n\n            l = loss(y_hat, y)\n\n            valid_loss = l.cpu().item()\n            # take max as the predicted label\n            valid_acc = (y_hat.argmax(dim=1) == y).sum().cpu().item()\n            valid_loss_list.append(valid_loss)\n            valid_acc_list.append(valid_acc)\n\n            valid_n += y.shape[0]\n\n        # paint\n        epoch_list.append(epoch + 1)\n        train_loss = sum(train_loss_list) / len(train_loss_list)\n        train_acc = sum(train_acc_list) / n\n        valid_loss = sum(valid_loss_list) / len(valid_loss_list)\n        valid_acc = sum(valid_acc_list) / valid_n\n        y_train_loss_list.append(train_loss)\n        y_valid_loss_list.append(valid_loss)\n        if epoch % 1 == 0 or epoch == num_epochs - 1:\n            plt.cla()\n            plt.title(\"Loss Curve\")\n            plt.xlabel(\"Epoch\")\n            plt.ylabel(\"Loss\")\n            plt.plot(epoch_list, y_train_loss_list)\n            plt.plot(epoch_list, y_valid_loss_list, color=\"orange\")\n            plt.legend([\"Train Loss\", \"Valid Loss\"])\n            # plt.scatter(epoch_list, y_train_loss_list, s=10)\n            # plt.scatter(epoch_list, y_valid_loss_list, s=10, color=\"orange\")\n            display.clear_output(wait=True)\n            plt.pause(0.00000001)\n\n        # output info\n        print(\n            'Epoch %d, Train Loss %.4f, Train Acc %.3f\\n%sValid Loss %.4f, Valid Acc %.3f' %\n            (epoch + 1,\n             train_loss, train_acc,\n             \" \" * len(\"Epoch %d, \" % (epoch + 1)),\n             valid_loss, valid_acc))\n\n    print(\"Training done\")\n\n    # save model\n    model_path = OUTPUT_PATH + \"model_{}.pth\".format(pd.Timestamp.now().strftime(\"%Y%m%d%H%M%S\"))\n    torch.save(model.state_dict(), model_path)\n    print(\"Model saved at\", model_path)\n\n    return model\n\n\nmodel = get_model(\"custom\")\n\nmodel = train(model, train_set, valid_set, batch_size=BATCH_SIZE,\n              num_epochs=NUM_EPOCHS, lr=LEARNING_RATE, weight_decay=WEIGHT_DECAY)","metadata":{},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## Predict","metadata":{}},{"cell_type":"code","source":"# model.load_state_dict(torch.load(\"model_20240110024206.pth\"))\n\ndef predict(model, test_set, device='auto'):\n    device = get_device(device)\n    \n    print(\"Predicting via %s...\" % device.type.upper())\n\n    test_iter = DataLoader(test_set, batch_size=1, num_workers=5, shuffle=False)\n\n    model = model.to(device)\n\n    model.eval()\n\n    pred_list = []\n\n    for X, _ in test_iter:\n        X = X.to(device)\n        y_hat = model(X)\n        pred_list.append(y_hat.argmax(dim=1).cpu().item())\n\n    print(\"Prediction done\")\n\n    return pred_list\n\n# Integrate prediction into dataset.img_labels (as int)\ndef integrate2dataset(dataset: LeafDataset, pred_list: list[int]):\n    dataset.img_labels = pred_list\n    return dataset\n\n# Convert type of dataset.img_labels to str\ndef get_str_labels(dataset: LeafDataset):\n    str_labels = dataset.img_labels[:]\n    if type(dataset.img_labels[0]) == int:\n        str_labels = [num2class[i] for i in dataset.img_labels]\n    return str_labels\n\ndef envelopeDataFrame(dataset: LeafDataset):\n    df = pd.DataFrame({\"image\": dataset.img_paths, \"label\": get_str_labels(dataset)})\n    return df\n\npred_list = predict(model, test_set)\nintegrate2dataset(test_set, pred_list)\nsubmission = envelopeDataFrame(test_set)\nsubmission.to_csv(OUTPUT_PATH + \"submission.csv\", index=False)\n\nsubmission","metadata":{},"execution_count":null,"outputs":[]}]}