{"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":"none","dataSources":[{"sourceId":23823,"databundleVersionId":1920183,"sourceType":"competition"},{"sourceId":1949522,"sourceType":"datasetVersion","datasetId":1157833},{"sourceId":11121490,"sourceType":"datasetVersion","datasetId":6935263}],"dockerImageVersionId":30918,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import os\nimport torch\nimport torch.nn as nn\nimport torch.nn.functional as F\nimport torch.optim as optim\nfrom torch.utils.data import Dataset, DataLoader, random_split\nimport torchvision\nimport torchvision.transforms as tfms\nimport numpy as np\nimport pandas as pd \nimport matplotlib.pyplot as plt \nimport seaborn as sns\nimport cv2 \nimport torch\nfrom PIL import Image\nfrom collections import defaultdict\nfrom tqdm import tqdm_notebook as tqdm\nimport sys; \nsys.path.insert(0,'../input/timm-nfnet')\nimport timm\nfrom sklearn.metrics import f1_score, average_precision_score\nimport random\n\n\nimport warnings\nwarnings.filterwarnings(\"ignore\")\n\nbase = \"/kaggle/input/hpa-single-cell-image-classification\"\n# base = \"/kaggle/input/hpa-single-small-dataset/\"\n\nTRAIN_DF_PATH = base + \"train.csv\"\nTRAIN_IMG_PATH = base + \"train\"\nTEST_IMG_PATH = base + \"test\"\nSAMPLE_SUB = base + \"sample_submission.csv\"\nCELL_LABEL = {\n0:  \"Nucleoplasm\", \n1:  \"Nuclear membrane\",   \n2:  \"Nucleoli\",   \n3:  \"Nucleoli fibrillar center\" ,  \n4:  \"Nuclear speckles\",\n5:  \"Nuclear bodies\",\n6:  \"Endoplasmic reticulum\",   \n7:  \"Golgi apparatus\",\n8:  \"Intermediate filaments\",\n9:  \"Actin filaments\", \n10: \"Microtubules\",\n11:  \"Mitotic spindle\",\n12:  \"Centrosome\",   \n13:  \"Plasma membrane\",\n14:  \"Mitochondria\",   \n15:  \"Aggresome\",\n16:  \"Cytosol\",   \n17:  \"Vesicles and punctate cytosolic patterns\",   \n18:  \"Negative\"\n}\n\ntrain_df = pd.read_csv(TRAIN_DF_PATH)\ntrain_df['label_count'] = train_df['Label'].apply(lambda x: len(x.split(\"|\")))\n# train_df.head()\n\n#hyperparameters\nCLASS = 19\nBATCH_SIZE = 32\nEPOCHS = 5\nLR = 1e-4\nRESIZE = 256\nDEVICE = torch.device('cuda') if torch.cuda.is_available() \\\n         else torch.device('cpu')\nPATH = base\nTRAIN_DIR = PATH + 'train/'\nTEST_DIR = PATH + 'test/'\n\n#imagenet transform\nimg_tfms = tfms.Compose([\n    tfms.ToPILImage(),\n    tfms.RandomHorizontalFlip(),\n    tfms.RandomVerticalFlip(),\n    tfms.RandomRotation(20),\n    tfms.RandomAffine(degrees=0, translate=(0.1, 0.1)),\n    tfms.ToTensor(),\n    tfms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))\n])\n\nDEVICE\n\nclass HPADataset(Dataset):\n    def __init__(self, csv_path, ids, labels=None, resize=None, transforms=None):\n        self.csv_path = csv_path\n        self.ids = ids\n        self.labels = labels  # Can be None for test set\n        self.resize = resize\n        self.transforms = transforms\n        \n    def __len__(self):\n        return len(self.ids)\n    \n    def __getitem__(self, index):\n        img_id = self.ids[index]\n        img_path = os.path.join(self.csv_path, img_id + '_green.png')\n        image = cv2.imread(img_path)\n\n        if image is None:\n            raise FileNotFoundError(f\"Image not found: {img_path}\")\n\n        if self.resize:\n            image = cv2.resize(image, (self.resize, self.resize))\n\n        # Ensure image is uint8 (for transforms compatibility)\n        if image.dtype != np.uint8:\n            image = (image * 255).astype(np.uint8) if image.max() <= 1.0 else image.astype(np.uint8)\n\n        if self.transforms:\n            image = self.transforms(image)\n\n        if self.labels is not None:  # Training/Validation mode\n            label = self.labels[index]\n            return image, torch.tensor(label, dtype=torch.float32)\n        else:  # Test mode\n            return image, img_id\n#model\nclass NFNet(nn.Module):\n    def __init__(self,output_features, model_name = 'nfnet_f1', pertrained=True):\n        super(NFNet, self).__init__()\n        self.model = timm.create_model(model_name, pretrained=pertrained)\n        self.model.head.fc = nn.Sequential(nn.Linear(self.model.head.fc.in_features, 512),\n                                 nn.ReLU(),\n                                 nn.Linear(512, output_features))\n        \n    def forward(self, x):\n        x = self.model(x)\n        return x\n\nclass CNNet(nn.Module):\n    def __init__(self, input_features, output_features):\n        super(CNNet, self).__init__()\n        self.model = torchvision.models.resnet34(pretrained=True)\n        self.model.fc = nn.Sequential(nn.Linear(input_features, 100),\n                                 nn.ReLU(),\n                                 nn.Linear(100, output_features))\n\n    def forward(self, x):\n        out = self.model(x)\n        return out\n\n\nclass FocalLoss(nn.Module):\n    def __init__(self, alpha=0.25, gamma=2.0, reduction='mean'):\n        super(FocalLoss, self).__init__()\n        self.alpha = alpha\n        self.gamma = gamma\n        self.reduction = reduction\n\n    def forward(self, inputs, targets):\n        # BCE with logits\n        bce_loss = F.binary_cross_entropy_with_logits(inputs, targets, reduction='none')\n        probas = torch.sigmoid(inputs)\n        pt = torch.where(targets == 1, probas, 1 - probas)\n        focal_term = self.alpha * (1 - pt) ** self.gamma\n        loss = focal_term * bce_loss\n\n        if self.reduction == 'mean':\n            return loss.mean()\n        elif self.reduction == 'sum':\n            return loss.sum()\n        else:\n            return loss\n\nimport random\n\nclass CutMixDataset(Dataset):\n    def __init__(self, dataset, num_classes=19, beta=1.0, prob=0.5):\n        self.dataset = dataset\n        self.num_classes = num_classes\n        self.beta = beta\n        self.prob = prob\n\n    def __len__(self):\n        return len(self.dataset)\n\n    def __getitem__(self, idx):\n        image1, label1 = self.dataset[idx]\n\n        # Apply CutMix with some probability\n        if random.random() < self.prob:\n            # Select random second sample\n            idx2 = random.randint(0, len(self.dataset) - 1)\n            image2, label2 = self.dataset[idx2]\n\n            lam = np.random.beta(self.beta, self.beta)\n            bbx1, bby1, bbx2, bby2 = self.rand_bbox(image1.size()[1:], lam)\n            image1[:, bby1:bby2, bbx1:bbx2] = image2[:, bby1:bby2, bbx1:bbx2]\n            lam = 1 - ((bbx2 - bbx1) * (bby2 - bby1) / (image1.size()[-1] * image1.size()[-2]))\n            label = label1 * lam + label2 * (1. - lam)\n            return image1, label\n\n        else:\n            return image1, label1\n\n    def rand_bbox(self, size, lam):\n        W = size[0]\n        H = size[1]\n        cut_rat = np.sqrt(1. - lam)\n        cut_w = int(W * cut_rat)\n        cut_h = int(H * cut_rat)\n\n        # uniform\n        cx = np.random.randint(W)\n        cy = np.random.randint(H)\n\n        bbx1 = np.clip(cx - cut_w // 2, 0, W)\n        bby1 = np.clip(cy - cut_h // 2, 0, H)\n        bbx2 = np.clip(cx + cut_w // 2, 0, W)\n        bby2 = np.clip(cy + cut_h // 2, 0, H)\n\n        return bbx1, bby1, bbx2, bby2\n\n\nprint(\"cell run\")\n","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2025-03-23T04:47:06.001139Z","iopub.execute_input":"2025-03-23T04:47:06.001567Z","iopub.status.idle":"2025-03-23T04:47:13.69042Z","shell.execute_reply.started":"2025-03-23T04:47:06.001534Z","shell.execute_reply":"2025-03-23T04:47:13.68949Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\nmodel = NFNet(CLASS)\nmodel = model.to(DEVICE)\nloss_fn = FocalLoss(alpha=0.25, gamma=2.0)\noptimizer = torch.optim.Adam(model.parameters(), lr=LR)\n\ntrain_df = pd.read_csv(PATH + \"train.csv\")\ntrain_df = train_df.sample(frac=1).reset_index(drop=True)  # Shuffle\nfrom sklearn.preprocessing import MultiLabelBinarizer\n\n# Split string labels into list of ints\ntrain_df['Label'] = train_df['Label'].apply(lambda x: list(map(int, x.split(\"|\"))))\n\n# Fit binarizer\nmlb = MultiLabelBinarizer(classes=range(CLASS))  # Ensure consistent order\ny_bin = mlb.fit_transform(train_df['Label'])\n\n# Then pass y_bin to your dataset\nX_train = train_df['ID'].values\nX_ds = HPADataset(TRAIN_DIR, X_train, y_bin, RESIZE, img_tfms)\n\n# 80-20 split\ntotal_size = len(X_ds)\nval_size = int(0.2 * total_size)\ntrain_size = total_size - val_size\ntrain_ds, valid_ds = random_split(X_ds, [train_size, val_size])\n\n\ncutmix_train_ds = CutMixDataset(train_ds, num_classes=CLASS, beta=1.0, prob=0.5)\ntrain_dl = DataLoader(cutmix_train_ds, batch_size=BATCH_SIZE, shuffle=True)\nvalid_dl = DataLoader(valid_ds, batch_size=BATCH_SIZE,shuffle=True)\n\nfrom sklearn.metrics import f1_score, average_precision_score, multilabel_confusion_matrix\nimport seaborn as sns\n\ndef evaluate_model(model, valid_dl, threshold=0.5, device=DEVICE):\n    model.eval()\n    all_preds, all_trues = [], []\n    images_sample, preds_sample, labels_sample = [], [], []\n\n    with torch.no_grad():\n        for img, lbl in valid_dl:\n            img = img.to(device)\n            lbl = lbl.to(device)\n            out = torch.sigmoid(model(img.float()))\n\n            all_preds.append(out.cpu().numpy())\n            all_trues.append(lbl.cpu().numpy())\n\n            if len(images_sample) < 8:  # Save samples for plotting\n                images_sample.extend(img.cpu().numpy())\n                preds_sample.extend(out.cpu().numpy())\n                labels_sample.extend(lbl.cpu().numpy())\n\n    preds = np.concatenate(all_preds)\n    trues = np.concatenate(all_trues)\n\n    preds_bin = (preds > threshold).astype(int)\n\n    f1 = f1_score(trues, preds_bin, average='macro')\n    map_score = average_precision_score(trues, preds, average='macro')\n    conf_matrix = multilabel_confusion_matrix(trues, preds_bin)\n\n    print(f\"F1 Score: {f1:.4f}\", end = \"\\t\")\n    print(f\"mAP: {map_score:.4f}\")\n\nprint(\"cell done\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-23T04:47:13.691617Z","iopub.execute_input":"2025-03-23T04:47:13.691892Z","iopub.status.idle":"2025-03-23T04:47:16.193656Z","shell.execute_reply.started":"2025-03-23T04:47:13.691869Z","shell.execute_reply":"2025-03-23T04:47:16.192561Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"#train loop\nloss_hist = []\nfor epoch in tqdm(range(EPOCHS)):\n    losses = []\n    model = model.train()\n    for batch_idx, (image, label) in enumerate(train_dl):\n        image = image.to(DEVICE)\n        label = label.to(DEVICE)\n        output = model(image.float())\n        loss = loss_fn(output, label)\n        losses.append(loss.item())\n        optimizer.zero_grad()\n        loss.backward()\n        optimizer.step()\n    loss_hist.append(sum(losses)/len(losses))\n    \n    print(f\"epoch: {epoch} loss:{sum(losses)/len(losses):.4f}\", end=\"\\t\")\n\n    evaluate_model(model, valid_dl)\n\n\nplt.figure(figsize=(15, 8))\nplt.title('Train Loss')\nplt.xlabel('epochs')\nplt.ylabel('loss')\nplt.plot(loss_hist)\nplt.show()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-03-23T04:48:05.747941Z","iopub.execute_input":"2025-03-23T04:48:05.748357Z","execution_failed":"2025-03-23T05:49:43.853Z"}},"outputs":[],"execution_count":null}]}