{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":65471,"databundleVersionId":7211464,"sourceType":"competition"}],"dockerImageVersionId":30636,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"### Datasets and DataLoaders","metadata":{}},{"cell_type":"code","source":"import os, torch, random, shutil, numpy as np\nfrom glob import glob; from PIL import Image\nfrom torch.utils.data import random_split, Dataset, DataLoader\nfrom torchvision import transforms as T\ntorch.manual_seed(2024)\n\nclass CustomDataset(Dataset):\n    \n    def __init__(self, root, data, transformations = None):\n        \n        self.data, self.transformations = data, transformations\n        self.im_paths = self.get_im_paths(root = root)\n        \n        self.cls_names, self.cls_counts, count, data_count = {}, {}, 0, 0\n        \n        if self.data == \"train\":\n            for idx, im_path in enumerate(self.im_paths):\n                class_name = self.get_class(im_path)\n                if class_name not in self.cls_names: self.cls_names[class_name] = count; self.cls_counts[class_name] = 1; count += 1\n                else: self.cls_counts[class_name] += 1\n        \n    def get_im_paths(self, root): return [im_path for im_path in sorted(glob(f\"{root}/{self.data}/*/*.png\"))] if self.data == \"train\" else [im_path for im_path in sorted(glob(f\"{root}/{self.data}/*.png\"))]\n    \n    def get_class(self, path): return os.path.basename(os.path.dirname(path))\n\n    def get_pos_neg_im_paths(self, qry_label):\n        \n        pos_im_paths = [im_path for im_path in self.im_paths if qry_label == self.get_class(im_path)]\n        neg_im_paths = [im_path for im_path in self.im_paths if qry_label != self.get_class(im_path)]\n        \n        pos_rand_int = random.randint(a = 0, b = len(pos_im_paths) - 1)\n        neg_rand_int = random.randint(a = 0, b = len(neg_im_paths) - 1)\n        \n        return pos_im_paths[pos_rand_int], neg_im_paths[neg_rand_int]\n    \n    def __len__(self): return len(self.im_paths)\n\n    def __getitem__(self, idx):\n        \n        if self.data == \"train\":\n            im_path = self.im_paths[idx]\n            qry_im = Image.open(im_path)\n            qry_label = self.get_class(im_path)\n\n            pos_im_path, neg_im_path = self.get_pos_neg_im_paths(qry_label = qry_label)\n            pos_im, neg_im = Image.open(pos_im_path), Image.open(neg_im_path)\n\n            qry_gt = self.cls_names[qry_label]\n            neg_gt = self.cls_names[self.get_class(neg_im_path)]\n\n            if self.transformations is not None: qry_im = self.transformations(qry_im); pos_im = self.transformations(pos_im); neg_im = self.transformations(neg_im)\n\n            data = {}\n\n            data[\"qry_im\"] = qry_im\n            data[\"qry_gt\"] = qry_gt\n            data[\"pos_im\"] = pos_im\n            data[\"neg_im\"] = neg_im\n            data[\"neg_gt\"] = neg_gt\n            \n            return data\n            \n        else: return self.transformations(Image.open(self.im_paths[idx]))\n    \ndef get_dls(root, transformations, bs, split = [0.9, 0.1], ns = 8):\n    \n    ds    = CustomDataset(root = root, data = \"train\", transformations = transformations)\n    ts_ds = CustomDataset(root = root, data = \"test\", transformations = transformations)\n    \n    total_len = len(ds); tr_len = int(total_len * split[0]); val_len = total_len - tr_len\n    \n    tr_ds, vl_ds, = random_split(dataset = ds, lengths = [tr_len, val_len])\n    tr_dl, val_dl, ts_dl = DataLoader(tr_ds, batch_size = bs, shuffle = True, num_workers = ns), DataLoader(vl_ds, batch_size = bs, shuffle = False, num_workers = ns), DataLoader(ts_ds, batch_size = 1, shuffle = False, num_workers = ns)\n    \n    return tr_dl, val_dl, ts_dl, ds.cls_names\n\nroot = \"/kaggle/input/arabic-letters-classification/Final_Arabic_Alpha_dataset/Final_Arabic_Alpha_dataset\"\nmean, std, im_size = [0.485, 0.456, 0.406], [0.229, 0.224, 0.225], 224\ntfs = T.Compose([T.Resize((im_size, im_size)), T.Grayscale(num_output_channels = 3), T.ToTensor(), T.Normalize(mean = mean, std = std)])\ntr_dl, val_dl, ts_dl, classes = get_dls(root = root, transformations = tfs, bs = 32)\n\nprint(len(tr_dl)); print(len(val_dl)); print(len(ts_dl)); print(classes)","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2024-01-17T04:23:06.490156Z","iopub.execute_input":"2024-01-17T04:23:06.490434Z","iopub.status.idle":"2024-01-17T04:23:13.757483Z","shell.execute_reply.started":"2024-01-17T04:23:06.490383Z","shell.execute_reply":"2024-01-17T04:23:13.756545Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Data Visualization","metadata":{}},{"cell_type":"code","source":"from matplotlib import pyplot as plt\n\ndef tensor_2_im(t, t_type = \"rgb\"):\n    \n    gray_tfs = T.Compose([T.Normalize(mean = [ 0.], std = [1/0.5]), T.Normalize(mean = [-0.5], std = [1])])\n    rgb_tfs = T.Compose([T.Normalize(mean = [ 0., 0., 0. ], std = [ 1/0.229, 1/0.224, 1/0.225 ]), T.Normalize(mean = [ -0.485, -0.456, -0.406 ], std = [ 1., 1., 1. ])])\n    \n    invTrans = gray_tfs if t_type == \"gray\" else rgb_tfs \n    \n    return (invTrans(t) * 255).detach().squeeze().cpu().permute(1,2,0).numpy().astype(np.uint8) if t_type == \"gray\" else (invTrans(t) * 255).detach().cpu().permute(1,2,0).numpy().astype(np.uint8)\n\ndef plot(rows, cols, count, im, title = \"Original Image\"):\n    \n    plt.subplot(rows, cols, count)\n    plt.imshow(tensor_2_im(im))\n    plt.axis(\"off\"); plt.title(title)\n    \n    return count + 1\n\ndef visualize(data, n_ims, data_type, cmap = None, cls_names = None):\n    \n    plt.figure(figsize = (25, 20))\n    rows = n_ims // 3; cols = n_ims // rows\n    count = 1\n    indices = [random.randint(0, len(data) - 1) for _ in range(n_ims)]\n    \n    for idx, index in enumerate(indices):\n        \n        if count == n_ims + 1: break\n        \n        if data_type in [\"train\", \"valid\"]:\n            meta_data = data[index]\n            qry_im, pos_im, neg_im, qry_lbl, neg_lbl = meta_data[\"qry_im\"], meta_data[\"pos_im\"], meta_data[\"neg_im\"], meta_data[\"qry_gt\"], meta_data[\"neg_gt\"]\n\n            # First Plot\n            count = plot(rows, cols, count, im = qry_im, title = f\"Query Image \\n Class -> {list(classes.keys())[qry_lbl]}\")\n\n            # Second Plot\n            count = plot(rows, cols, count, im = pos_im, title = f\"Positive Image \\n Class -> {list(classes.keys())[qry_lbl]}\")\n\n            # Third Plot\n            count = plot(rows, cols, count, im = neg_im, title = f\"Negative Image \\n Class -> {list(classes.keys())[neg_lbl]}\")\n            \n        else:\n            # First Plot\n            count = plot(rows, cols, count, im = data[index], title = \"Test Image has no class\")\n        \nvisualize(data = tr_dl.dataset, data_type = \"train\", n_ims = 18)","metadata":{"execution":{"iopub.status.busy":"2024-01-17T04:23:13.759071Z","iopub.execute_input":"2024-01-17T04:23:13.759481Z","iopub.status.idle":"2024-01-17T04:23:17.300854Z","shell.execute_reply.started":"2024-01-17T04:23:13.759453Z","shell.execute_reply":"2024-01-17T04:23:17.299984Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"visualize(data = val_dl.dataset, data_type = \"valid\", n_ims = 18)","metadata":{"execution":{"iopub.status.busy":"2024-01-17T04:23:17.301932Z","iopub.execute_input":"2024-01-17T04:23:17.302186Z","iopub.status.idle":"2024-01-17T04:23:20.69683Z","shell.execute_reply.started":"2024-01-17T04:23:17.302163Z","shell.execute_reply":"2024-01-17T04:23:20.695927Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"visualize(data = ts_dl.dataset, data_type = \"test\", n_ims = 18)","metadata":{"execution":{"iopub.status.busy":"2024-01-17T04:23:20.699334Z","iopub.execute_input":"2024-01-17T04:23:20.700177Z","iopub.status.idle":"2024-01-17T04:23:22.460212Z","shell.execute_reply.started":"2024-01-17T04:23:20.700135Z","shell.execute_reply":"2024-01-17T04:23:22.459327Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Data Analysis","metadata":{}},{"cell_type":"code","source":"def data_analysis(root, data, transformations):\n    \n    ds = CustomDataset(root = root, data = data, transformations = transformations)\n    cls_counts, width, text_width = ds.cls_counts,  0.7, 0.05\n    text_height = 2\n    cls_names = list(cls_counts.keys()); counts = list(cls_counts.values())\n    \n    fig, ax = plt.subplots(figsize = (20, 10))\n    indices = np.arange(len(counts))\n\n    ax.bar(indices, counts, width, color = \"firebrick\")\n    ax.set_xlabel(\"Class Names\", color = \"red\")\n    ax.set_xticklabels(cls_names, rotation = 60)\n    ax.set(xticks = indices, xticklabels = cls_names)\n    ax.set_ylabel(\"Data Counts\", color = \"red\")\n    ax.set_title(f\"Dataset Class Imbalance Analysis\")\n\n    for i, v in enumerate(counts): ax.text(i - text_width, v + text_height, str(v), color = \"royalblue\")\n    \ndata_analysis(root = root, data = \"train\", transformations = tfs)","metadata":{"execution":{"iopub.status.busy":"2024-01-17T04:23:22.461384Z","iopub.execute_input":"2024-01-17T04:23:22.461665Z","iopub.status.idle":"2024-01-17T04:23:23.816851Z","shell.execute_reply.started":"2024-01-17T04:23:22.46164Z","shell.execute_reply":"2024-01-17T04:23:23.815917Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### ","metadata":{}},{"cell_type":"markdown","source":"### Train and Validation","metadata":{}},{"cell_type":"code","source":"import timm, torchmetrics\nfrom tqdm import tqdm\nm = timm.create_model(\"rexnet_150\", pretrained = True, num_classes = len(classes))  \ndef train_setup(m): return m.to(\"cuda\").eval(), 5, \"cuda\", [torch.nn.CrossEntropyLoss(), torch.nn.CosineEmbeddingLoss()], torch.optim.Adam(params = m.parameters(), lr = 3e-4)\ndef to_device(batch, device): return batch[\"qry_im\"].to(device), batch[\"qry_gt\"].to(device), batch[\"pos_im\"].to(device), batch[\"neg_im\"].to(device), batch[\"neg_gt\"].to(device)\ndef get_acc(preds, gts): return (torch.argmax(preds, dim = 1) == gts).sum().item()\ndef get_batch_acc(qry_preds, pos_preds, qry_gts): return (get_acc(preds = qry_preds, gts = qry_gts) + get_acc(preds = pos_preds, gts = qry_gts)) / 2\ndef get_batch_f1(qry_preds, pos_preds, qry_gts): return (f1_score(qry_preds, qry_gts).item() + f1_score(pos_preds, qry_gts).item()) / 2\n\ndef get_fm(fm):\n        \n        \"\"\"\n        \n        This function gets feature map with size (bs, fm_shape, 7, 7)\n        applies average pooling and returns feature map with shape (bs, fm_shape).\n        \n        Parameter:\n        \n            fm - feature map, tensor.\n        \n        Output:\n        \n            fm - reshaped feature map, tensor.\n        \n        \"\"\"\n        \n        pool = torch.nn.AvgPool2d((fm.shape[2],fm.shape[3]))\n        \n        return torch.reshape(pool(fm), (-1, fm.shape[1]))\n\nm, epochs, device, loss_fns, optimizer = train_setup(m)\nf1_score = torchmetrics.F1Score(task = \"multiclass\", num_classes = len(classes)).to(device)\n\nsave_prefix, save_dir = \"arabic\", \"saved_models\"\nprint(\"Start training...\")\nbest_acc, best_loss, threshold, not_improved, patience = 0, float(\"inf\"), 0.01, 0, 3\ntr_losses, val_losses, tr_accs, val_accs, tr_f1s, val_f1s = [], [], [], [], [], []\ncos_lbls = {\"cos_pos\": torch.tensor(1.).unsqueeze(0), \"cos_neg\": torch.tensor(-1.).unsqueeze(0)}\n\nbest_loss = float(torch.inf)\nce_loss_fn, cos_loss_fn = loss_fns[0], loss_fns[1]\n    \nfor epoch in range(epochs):\n\n    epoch_loss, epoch_acc, epoch_f1 = 0, 0, 0\n    for idx, batch in tqdm(enumerate(tr_dl)):\n        qry_ims, qry_gts, pos_ims, neg_ims, neg_gts = to_device(batch, device)\n        \n        # Get feature maps before pooling\n        qry_fms = m.forward_features(qry_ims)\n        pos_fms = m.forward_features(pos_ims)\n        neg_fms = m.forward_features(neg_ims)\n        \n        # Get predicted logits\n        qry_preds = m.forward_head(qry_fms)\n        pos_preds = m.forward_head(pos_fms)\n        neg_preds = m.forward_head(neg_fms)\n        \n        # Extract feature maps\n        qry_fms, pos_fms, neg_fms = get_fm(qry_fms), get_fm(pos_fms), get_fm(neg_fms)\n\n        # Compute loss scores\n        # a) compute cosine embedding loss\n        cos_loss = cos_loss_fn(qry_fms, pos_fms, cos_lbls[\"cos_pos\"].to(device)) + cos_loss_fn(qry_fms, neg_fms, cos_lbls[\"cos_neg\"].to(device))\n\n        # b) compute cross entropy loss\n        ce_loss = ce_loss_fn(qry_preds, qry_gts) + ce_loss_fn(pos_preds, qry_gts)\n\n        loss = cos_loss + ce_loss\n        epoch_loss += loss.item()\n        \n        # Compute accuracy scores\n        epoch_acc += get_batch_acc(qry_preds, pos_preds, qry_gts)\n        epoch_f1  += get_batch_f1(qry_preds, pos_preds, qry_gts)\n\n        optimizer.zero_grad(); loss.backward(); optimizer.step()\n\n    tr_loss_to_track = epoch_loss / len(tr_dl)\n    tr_f1_to_track   = epoch_f1   / len(tr_dl)\n    tr_acc_to_track  = epoch_acc  / len(tr_dl.dataset)\n    tr_losses.append(tr_loss_to_track); tr_accs.append(tr_acc_to_track); tr_f1s.append(tr_f1_to_track)\n\n    print(f\"{epoch + 1}-epoch train process is completed!\")\n    print(f\"{epoch + 1}-epoch train loss          -> {tr_loss_to_track:.3f}\")\n    print(f\"{epoch + 1}-epoch train accuracy      -> {tr_acc_to_track:.3f}\")\n    print(f\"{epoch + 1}-epoch train f1-score      -> {tr_f1_to_track:.3f}\")\n\n    m.eval()\n    with torch.no_grad():\n        val_epoch_loss, val_epoch_acc, val_epoch_f1 = 0, 0, 0\n        for idx, batch in enumerate(val_dl):\n            \n            qry_ims, qry_gts, pos_ims, neg_ims, neg_gts = to_device(batch, device)\n        \n            # Get feature maps before pooling\n            qry_fms, pos_fms, neg_fms = m.forward_features(qry_ims), m.forward_features(pos_ims), m.forward_features(neg_ims)\n\n            # Get predicted logits\n            qry_preds, pos_preds, neg_preds = m.forward_head(qry_fms), m.forward_head(pos_fms), m.forward_head(neg_fms)\n\n            # Extract feature maps\n            qry_fms, pos_fms, neg_fms = get_fm(qry_fms), get_fm(pos_fms), get_fm(neg_fms)\n\n            # Compute loss scores\n            # a) compute cosine embedding loss\n            cos_loss = cos_loss_fn(qry_fms, pos_fms, cos_lbls[\"cos_pos\"].to(device)) + cos_loss_fn(qry_fms, neg_fms, cos_lbls[\"cos_neg\"].to(device))\n\n            # b) compute cross entropy loss\n            ce_loss = ce_loss_fn(qry_preds, qry_gts) + ce_loss_fn(pos_preds, qry_gts)\n\n            loss = cos_loss + ce_loss\n\n            # Compute accuracy scores\n            val_epoch_acc += get_batch_acc(qry_preds, pos_preds, qry_gts)\n            val_epoch_f1  += get_batch_f1(qry_preds, pos_preds, qry_gts)\n            val_epoch_loss += loss.item()\n\n        val_loss_to_track = val_epoch_loss / len(val_dl)\n        val_f1_to_track   = val_epoch_f1   / len(val_dl)\n        val_acc_to_track  = val_epoch_acc  / len(val_dl.dataset)\n        val_losses.append(val_loss_to_track); val_accs.append(val_acc_to_track); val_f1s.append(val_f1_to_track)\n\n        print(f\"{epoch + 1}-epoch validation process is completed!\")\n        print(f\"{epoch + 1}-epoch validation loss     -> {val_loss_to_track:.3f}\")\n        print(f\"{epoch + 1}-epoch validation accuracy -> {val_acc_to_track:.3f}\")\n        print(f\"{epoch + 1}-epoch validation f1-score -> {val_f1_to_track:.3f}\")\n\n        if val_loss_to_track < (best_loss + threshold):\n            os.makedirs(save_dir, exist_ok = True)\n            best_loss = val_loss_to_track\n            torch.save(m.state_dict(), f\"{save_dir}/{save_prefix}_best_model.pth\")\n            \n        else:\n            not_improved += 1\n            print(f\"Loss value did not decrease for {not_improved} epochs\")\n            if not_improved == patience:\n                print(f\"Stop training since loss value did not decrease for {patience} epochs.\")\n                break","metadata":{"execution":{"iopub.status.busy":"2024-01-17T04:23:23.818149Z","iopub.execute_input":"2024-01-17T04:23:23.818476Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Learning Curves","metadata":{}},{"cell_type":"code","source":"def learning_curves(tr_losses, val_losses, tr_accs, val_accs, tr_f1s, val_f1s):\n    \n    plt.figure(figsize = (10, 5))\n    plt.plot(tr_losses, label = \"Train Loss\", c = \"red\")\n    plt.plot(val_losses, label = \"Validation Loss\", c = \"blue\")\n    plt.xlabel(\"Epochs\"); plt.ylabel(\"Loss Values\")\n    plt.xticks(ticks = np.arange(len(tr_losses)), labels = [i for i in range(1, len(tr_losses) + 1)])\n    plt.legend(); plt.show()\n    \n    plt.figure(figsize = (10, 5))\n    plt.plot(tr_accs, label = \"Train Accuracy\", c = \"orangered\")\n    plt.plot(val_accs, label = \"Validation Accuracy\", c = \"darkgreen\")\n    plt.xlabel(\"Epochs\"); plt.ylabel(\"Accuracy Scores\")\n    plt.xticks(ticks = np.arange(len(tr_accs)), labels = [i for i in range(1, len(tr_accs) + 1)])\n    plt.legend(); plt.show()\n    \n    plt.figure(figsize = (10, 5))\n    plt.plot([tr_f1.cpu() for tr_f1 in tr_f1s], label = \"Train F1 Score\", c = \"orangered\")\n    plt.plot([val_f1.cpu() for val_f1 in val_f1s], label = \"Validation F1 Score\", c = \"darkgreen\")\n    plt.xlabel(\"Epochs\"); plt.ylabel(\"F1 Scores\")\n    plt.xticks(ticks = np.arange(len(tr_f1s)), labels = [i for i in range(1, len(tr_f1s) + 1)])\n    plt.legend(); plt.show()\n    \nlearning_curves(tr_losses, val_losses, tr_accs, val_accs, tr_f1s, val_f1s)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"### Inference and Model Performance Analysis","metadata":{}},{"cell_type":"code","source":"import cv2\nclass SaveFeatures():\n    \n    \"\"\" Extract pretrained activations\"\"\"\n    features = None\n    def __init__(self, m):\n        self.hook = m.register_forward_hook(self.hook_fn)\n    def hook_fn(self, module, input, output):\n        self.features = ((output.cpu()).data).numpy()\n    def remove(self): self.hook.remove()\n\ndef getCAM(conv_fs, linear_weights, class_idx):\n    \n    bs, chs, h, w = conv_fs.shape\n    cam = linear_weights[class_idx].dot(conv_fs[0,:, :, ].reshape((chs, h * w)))\n    cam = cam.reshape(h, w)\n    \n    return (cam - np.min(cam)) / np.max(cam)\n\ndef inference(model, device, test_dl, num_ims, row, final_conv, fc_params, cls_names = None):\n    \n    weight, acc = np.squeeze(fc_params[0].cpu().data.numpy()), 0\n    activated_features = SaveFeatures(final_conv)\n    preds, images = [], []\n    for idx, batch in tqdm(enumerate(test_dl)):\n        im = batch.to(device)\n        pred_class = torch.argmax(model(im), dim = 1)\n        images.append(im)\n        preds.append(pred_class.item())\n    \n    plt.figure(figsize = (20, 10))\n    indekslar = [random.randint(0, len(images) - 1) for _ in range(num_ims)]\n    \n    for idx, indeks in enumerate(indekslar):\n        \n        im = images[indeks].squeeze()\n        pred_idx = preds[indeks]\n        heatmap = getCAM(activated_features.features, weight, pred_idx)\n        \n        # Start plot\n        plt.subplot(row, num_ims // row, idx + 1)\n        plt.imshow(tensor_2_im(im), cmap = \"gray\"); plt.axis(\"off\")\n        plt.imshow(cv2.resize(heatmap, (im_size, im_size), interpolation=cv2.INTER_LINEAR), alpha=0.4, cmap='jet'); plt.axis(\"off\")\n        \n        if cls_names is not None: plt.title(f\"PRED -> {cls_names[int(preds[indeks])]}\")\n        else: plt.title(f\"GT -> {gt} ; PRED -> {pred}\")\n\nm.load_state_dict(torch.load(f\"{save_dir}/{save_prefix}_best_model.pth\"))\nm.eval()\nfinal_conv, fc_params = m.features[-1], list(m.head.fc.parameters())\ninference(model = m.to(device), device = device, test_dl = ts_dl, num_ims = 20, row = 4, cls_names = list(classes.keys()), final_conv = final_conv, fc_params = fc_params)","metadata":{"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}