{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.11.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":22881,"databundleVersionId":1552852,"sourceType":"competition"},{"sourceId":7385384,"sourceType":"datasetVersion","datasetId":4292647},{"sourceId":647199,"sourceType":"modelInstanceVersion","isSourceIdPinned":true,"modelInstanceId":488099,"modelId":503524}],"dockerImageVersionId":31154,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Softmax (Supervised Leaning)","metadata":{}},{"cell_type":"code","source":"# ============================================================\n# MobileFaceNet — Face Verification (from scratch, PyTorch)\n# ============================================================\n\nimport os\nimport torch\nfrom torch import nn, optim\nfrom torch.utils.data import DataLoader\nfrom torchvision import datasets, transforms\nimport torch.nn.functional as F\n\n# -----------------------------\n# 1. CONFIG\n# -----------------------------\nDATA_ROOT = \"/kaggle/input/11-785-fall-20-homework-2-part-2\"\nCLS_ROOT  = os.path.join(DATA_ROOT, \"classification_data\")\nTRAIN_DIR = os.path.join(CLS_ROOT, \"train_data\")\nVAL_DIR   = os.path.join(CLS_ROOT, \"val_data\")\nTEST_DIR  = os.path.join(CLS_ROOT, \"test_data\")\n\nIMG_SIZE   = 80   \nBATCH_SIZE = 64\nEMBED_DIM  = 128\n\ndevice = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\nprint(\"Device:\", device)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-15T03:20:57.539753Z","iopub.execute_input":"2025-11-15T03:20:57.540427Z","iopub.status.idle":"2025-11-15T03:21:04.090195Z","shell.execute_reply.started":"2025-11-15T03:20:57.540398Z","shell.execute_reply":"2025-11-15T03:21:04.089270Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -----------------------------\n# 2. DATASET & DATALOADER\n# -----------------------------\ntorch.backends.cudnn.benchmark = True  # ✅ optimize conv performance\nfrom torchvision.transforms import InterpolationMode\n\ntrain_transform = transforms.Compose([\n    transforms.Resize((IMG_SIZE, IMG_SIZE), interpolation=InterpolationMode.BILINEAR),\n    transforms.RandomHorizontalFlip(),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.5, 0.5, 0.5],\n                         std=[0.5, 0.5, 0.5])\n])\n\nval_transform = transforms.Compose([\n    transforms.Resize((IMG_SIZE, IMG_SIZE), interpolation=InterpolationMode.BILINEAR),\n    transforms.ToTensor(),\n    transforms.Normalize(mean=[0.5, 0.5, 0.5],\n                         std=[0.5, 0.5, 0.5])\n])\n\ntrain_dataset = datasets.ImageFolder(TRAIN_DIR, transform=train_transform)\nval_dataset   = datasets.ImageFolder(VAL_DIR,   transform=val_transform)\ntest_dataset  = datasets.ImageFolder(TEST_DIR,  transform=val_transform)\n\n# ✅ Optimized DataLoader setup\ntrain_loader = DataLoader(train_dataset, batch_size=64,\n                          shuffle=True, num_workers=2,\n                          pin_memory=True, prefetch_factor=4)\nval_loader   = DataLoader(val_dataset, batch_size=64,\n                          shuffle=False, num_workers=2,\n                          pin_memory=True, prefetch_factor=4)\ntest_loader  = DataLoader(test_dataset, batch_size=64,\n                          shuffle=False, num_workers=2,\n                          pin_memory=True, prefetch_factor=4)\n\n\nnum_classes = len(train_dataset.classes)\nprint(\"Num classes:\", num_classes)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-15T03:21:04.092059Z","iopub.execute_input":"2025-11-15T03:21:04.092382Z","iopub.status.idle":"2025-11-15T03:28:01.731246Z","shell.execute_reply.started":"2025-11-15T03:21:04.092363Z","shell.execute_reply":"2025-11-15T03:28:01.730528Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -----------------------------\n# 3. MODEL — MobileFaceNet architecture\n# -----------------------------\n\n# --- Helper blocks ---\ndef conv_bn(inp, oup, stride):\n    \"\"\"Standard 3x3 Conv + BN + PReLU\"\"\"\n    return nn.Sequential(\n        nn.Conv2d(inp, oup, 3, stride, 1, bias=False),\n        nn.BatchNorm2d(oup),\n        nn.PReLU(oup)\n    )\n\ndef conv_dw(inp, oup, stride):\n    \"\"\"Depthwise + Pointwise conv (main efficiency trick)\"\"\"\n    return nn.Sequential(\n        nn.Conv2d(inp, inp, 3, stride, 1, groups=inp, bias=False),\n        nn.BatchNorm2d(inp),\n        nn.PReLU(inp),\n        nn.Conv2d(inp, oup, 1, 1, 0, bias=False),\n        nn.BatchNorm2d(oup),\n        nn.PReLU(oup)\n    )\n\nclass Bottleneck(nn.Module):\n    \"\"\"Inverted residual block (from MobileNetV2)\"\"\"\n    def __init__(self, inp, oup, stride, expand_ratio):\n        super().__init__()\n        self.stride = stride\n        hidden_dim = int(round(inp * expand_ratio))\n        self.use_res_connect = self.stride == 1 and inp == oup\n\n        layers = []\n        if expand_ratio != 1:\n            # 1x1 pointwise expand\n            layers.extend([\n                nn.Conv2d(inp, hidden_dim, 1, 1, 0, bias=False),\n                nn.BatchNorm2d(hidden_dim),\n                nn.PReLU(hidden_dim)\n            ])\n        # 3x3 depthwise conv + 1x1 projection\n        layers.extend([\n            nn.Conv2d(hidden_dim, hidden_dim, 3, stride, 1,\n                      groups=hidden_dim, bias=False),\n            nn.BatchNorm2d(hidden_dim),\n            nn.PReLU(hidden_dim),\n            nn.Conv2d(hidden_dim, oup, 1, 1, 0, bias=False),\n            nn.BatchNorm2d(oup)\n        ])\n        self.conv = nn.Sequential(*layers)\n        self.prelu = nn.PReLU(oup)\n\n    def forward(self, x):\n        if self.use_res_connect:\n            return x + self.prelu(self.conv(x))\n        else:\n            return self.prelu(self.conv(x))\n\nclass MobileFaceNet(nn.Module):\n    \"\"\"Main CNN model — extracts 128-D normalized embeddings\"\"\"\n    def __init__(self, embedding_size=128, num_classes=None):\n        super().__init__()\n        self.features = nn.Sequential(\n            conv_bn(3, 64, 2),\n            conv_dw(64, 64, 1),\n            Bottleneck(64, 64, 2, 2),\n            Bottleneck(64, 64, 1, 2),\n            Bottleneck(64, 128, 2, 4),\n            Bottleneck(128, 128, 1, 2),\n            Bottleneck(128, 128, 1, 4),\n            Bottleneck(128, 128, 1, 2),\n            conv_bn(128, 512, 1)\n        )\n        # Global depthwise conv\n        # Use adaptive pooling instead of fixed 7x7 kernel for flexible input (e.g., 64x64)\n        self.depthwise = nn.Conv2d(512, 512, kernel_size=3, stride=1, padding=1, groups=512, bias=False)\n        self.depthwise_bn = nn.BatchNorm2d(512)\n        self.pool = nn.AdaptiveAvgPool2d((1, 1))\n        self.linear = nn.Linear(512, embedding_size)\n        self.bn = nn.BatchNorm1d(embedding_size)\n\n        # Optional classification head\n        self.classifier = nn.Linear(embedding_size, num_classes) if num_classes else None\n\n    def forward(self, x, return_embedding=False):\n        \"\"\"Forward pass:\n        - Return logits if training (classification)\n        - Return L2-normalized embedding if inference\n        \"\"\"\n        x = self.features(x)\n        x = self.depthwise(x)\n        x = self.depthwise_bn(x)\n        x = self.pool(x)\n        x = x.view(x.size(0), -1)          # flatten\n        emb = self.linear(x)               # project to 128D\n        emb = self.bn(emb)\n        emb = F.normalize(emb, p=2, dim=1) # L2 normalization\n\n        if return_embedding or self.classifier is None:\n            return emb                     # inference mode\n        else:\n            logits = self.classifier(emb)  # training mode\n            return logits","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-15T03:28:01.731952Z","iopub.execute_input":"2025-11-15T03:28:01.732183Z","iopub.status.idle":"2025-11-15T03:28:01.744766Z","shell.execute_reply.started":"2025-11-15T03:28:01.732156Z","shell.execute_reply":"2025-11-15T03:28:01.744181Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -----------------------------\n# 4. TRAINING LOOP\n# -----------------------------\nfrom tqdm.notebook import tqdm\n\nmodel = MobileFaceNet(embedding_size=EMBED_DIM, num_classes=num_classes).to(device)\ncriterion = nn.CrossEntropyLoss()\noptimizer = torch.optim.AdamW(\n    model.parameters(),\n    lr=1e-3,\n    weight_decay=1e-4\n)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-15T03:28:01.745628Z","iopub.execute_input":"2025-11-15T03:28:01.745836Z","iopub.status.idle":"2025-11-15T03:28:02.096973Z","shell.execute_reply.started":"2025-11-15T03:28:01.745821Z","shell.execute_reply":"2025-11-15T03:28:02.096348Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# -----------------------------\n# 5. TRAINING STAGE (with early stopping)\n\nEPOCHS = 30\nbest_val_acc = 0.0\nbest_val_loss = float('inf')\npatience = 5           # stop if no improvement after 5 epochs\nwait = 0\nsave_path = \"mobilefacenet_best.pth\"\n\nfrom torch.cuda.amp import autocast, GradScaler\n\nscaler = GradScaler()  # scales the loss for stability\n\nfor epoch in range(1, EPOCHS + 1):\n    model.train()\n    running_loss, correct, total = 0.0, 0, 0\n\n    pbar = tqdm(train_loader, desc=f\"Training Epoch {epoch}\", leave=False)\n\n    for imgs, labels in pbar:\n        imgs, labels = imgs.to(device, non_blocking=True), labels.to(device, non_blocking=True)\n        optimizer.zero_grad()\n\n        # ----------------------------\n        # ✅ Mixed-Precision Training\n        # ----------------------------\n        with autocast():  # automatically choose float16/float32 per op\n            outputs = model(imgs)\n            loss = criterion(outputs, labels)\n\n        # backward with scaled loss\n        scaler.scale(loss).backward()\n        scaler.step(optimizer)\n        scaler.update()\n\n        # metrics\n        _, preds = torch.max(outputs, 1)\n        running_loss += loss.item() * imgs.size(0)\n        correct += (preds == labels).sum().item()\n        total += imgs.size(0)\n\n        pbar.set_postfix(loss=loss.item())\n\n    train_loss = running_loss / total\n    train_acc = correct / total\n\n    # ---- VALIDATION ----\n    model.eval()\n    val_loss, val_correct, val_total = 0.0, 0, 0\n\n    with torch.no_grad():\n        for imgs, labels in val_loader:\n            imgs, labels = imgs.to(device, non_blocking=True), labels.to(device, non_blocking=True)\n            outputs = model(imgs)\n            loss = criterion(outputs, labels)\n            _, preds = torch.max(outputs, 1)\n\n            val_loss += loss.item() * imgs.size(0)\n            val_correct += (preds == labels).sum().item()\n            val_total += imgs.size(0)\n\n    val_loss /= val_total\n    val_acc = val_correct / val_total\n\n    print(f\"[Epoch {epoch:02d}/{EPOCHS}] \"\n          f\"Train Loss: {train_loss:.4f} | Acc: {train_acc:.4f} \"\n          f\"|| Val Loss: {val_loss:.4f} | Acc: {val_acc:.4f}\")\n\n    # ---- EARLY STOPPING LOGIC ----\n    if val_acc > best_val_acc:\n        best_val_acc = val_acc\n        best_val_loss = val_loss\n        torch.save(model.state_dict(), save_path)\n        print(f\"✅ New best model saved at epoch {epoch} (Val Acc: {val_acc:.4f})\")\n        wait = 0\n    else:\n        wait += 1\n        print(f\"No improvement. Patience: {wait}/{patience}\")\n\n        if wait >= patience:\n            print(f\"⏹️ Early stopping triggered at epoch {epoch}\")\n            break\n\nprint(f\"\\nTraining complete. Best Val Acc: {best_val_acc:.4f}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-15T03:57:03.572791Z","iopub.execute_input":"2025-11-15T03:57:03.573465Z","execution_failed":"2025-11-15T06:53:29.918Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Triplet Loss (Metric Learning)","metadata":{}},{"cell_type":"code","source":"# ===========================\n# SIAMESE NETWORK - FACE VERIFICATION\n# ===========================\nimport os, random, torch, torch.nn as nn, torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader\nfrom torchvision import transforms\nfrom sklearn.metrics import roc_auc_score, roc_curve\nfrom tqdm import tqdm\nimport numpy as np\nfrom PIL import Image\nfrom torch.cuda.amp import GradScaler, autocast\n\nDEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\nprint('Device:', DEVICE)\n\n# ===========================\n# CONFIG\n# ===========================\nCFG = {\n    'data_root': '/kaggle/input/11-785-fall-20-homework-2-part-2/classification_data/train_data',\n    'ver_pairs': '/kaggle/input/11-785-fall-20-homework-2-part-2/verification_pairs_val.txt',\n    'ver_root': '/kaggle/input/11-785-fall-20-homework-2-part-2/',\n    'img_size': 80,\n    'emb_dim': 128,\n    'lr': 3e-4,\n    'batch_size': 128,\n    'epochs': 10,\n    'num_workers': 2,\n    'save_path': '/kaggle/working/siamese_model.pth'\n}","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ===========================\n# MODEL - MOBILEFACENET\n# ===========================\ndef conv_bn(inp, oup, stride=1):\n    return nn.Sequential(\n        nn.Conv2d(inp, oup, 3, stride, 1, bias=False),\n        nn.BatchNorm2d(oup),\n        nn.PReLU()\n    )\n\ndef conv_dw(inp, oup, stride=1):\n    return nn.Sequential(\n        nn.Conv2d(inp, inp, 3, stride, 1, groups=inp, bias=False),\n        nn.BatchNorm2d(inp),\n        nn.PReLU(),\n        nn.Conv2d(inp, oup, 1, 1, 0, bias=False),\n        nn.BatchNorm2d(oup),\n        nn.PReLU()\n    )\n\nclass MobileFaceNet(nn.Module):\n    def __init__(self, embedding_dim=256):\n        super().__init__()\n        self.conv1 = conv_bn(3, 64, 2)\n        self.conv2 = conv_dw(64, 64, 1)\n        self.conv3 = conv_dw(64, 128, 2)\n        self.conv4 = conv_dw(128, 128, 1)\n        self.conv5 = conv_dw(128, 256, 2)\n        self.conv6 = conv_dw(256, 256, 1)\n        self.conv7 = conv_dw(256, 512, 2)\n        self.conv8 = nn.Sequential(\n            conv_dw(512, 512, 1),\n            conv_dw(512, 512, 1),\n            conv_dw(512, 512, 1),\n            conv_dw(512, 512, 1),\n            conv_dw(512, 512, 1),\n        )\n        self.conv9 = conv_dw(512, 512, 1)\n        self.conv10 = conv_dw(512, 1024, 1)\n        self.avg = nn.AdaptiveAvgPool2d(1)\n        self.fc = nn.Sequential(\n            nn.Dropout(0.2),\n            nn.Linear(1024, embedding_dim),\n            nn.BatchNorm1d(embedding_dim)\n        )\n    \n    def forward(self, x):\n        x = self.conv1(x)\n        x = self.conv2(x)\n        x = self.conv3(x)\n        x = self.conv4(x)\n        x = self.conv5(x)\n        x = self.conv6(x)\n        x = self.conv7(x)\n        x = self.conv8(x)\n        x = self.conv9(x)\n        x = self.conv10(x)\n        x = self.avg(x).view(x.size(0), -1)\n        x = self.fc(x)\n        return F.normalize(x, p=2, dim=1)\n\n# ===========================\n# DATASET - VALIDATION\n# ===========================\nclass VerificationDataset(Dataset):\n    def __init__(self, img_paths, root):\n        self.img_paths = img_paths\n        self.root = root\n        self.tfms = transforms.Compose([\n            transforms.Resize((CFG['img_size'], CFG['img_size'])),\n            transforms.ToTensor(),\n            transforms.Normalize([0.5]*3, [0.5]*3)\n        ])\n    \n    def __len__(self):\n        return len(self.img_paths)\n    \n    def __getitem__(self, idx):\n        path = os.path.join(self.root, self.img_paths[idx])\n        try:\n            img = Image.open(path).convert('RGB')\n            return self.tfms(img)\n        except:\n            return torch.zeros(3, CFG['img_size'], CFG['img_size'])\n","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ===========================\n# SIAMESE MODEL\n# ===========================\nclass SiameseNetwork(nn.Module):\n    def __init__(self, embedding_dim=256):\n        super().__init__()\n        self.backbone = MobileFaceNet(embedding_dim)\n    \n    def forward(self, img1, img2):\n        emb1 = self.backbone(img1)\n        emb2 = self.backbone(img2)\n        return emb1, emb2\n    \n    def get_embedding(self, img):\n        return self.backbone(img)\n\n# ===========================\n# CONTRASTIVE LOSS\n# ===========================\nclass ContrastiveLoss(nn.Module):\n    def __init__(self, margin=1.0):\n        super().__init__()\n        self.margin = margin\n    \n    def forward(self, emb1, emb2, target):\n        # Euclidean distance\n        dist = F.pairwise_distance(emb1, emb2)\n        \n        # Contrastive loss\n        loss_pos = target * dist.pow(2)\n        loss_neg = (1 - target) * F.relu(self.margin - dist).pow(2)\n        loss = (loss_pos + loss_neg).mean()\n        \n        return loss","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ===========================\n# TRAINING\n# ===========================\ndef train_one_epoch(model, loader, optimizer, criterion, scaler, epoch):\n    model.train()\n    running_loss = 0.\n    pbar = tqdm(loader, desc=f\"Epoch {epoch}\")\n    \n    for img1, img2, target in pbar:\n        img1 = img1.to(DEVICE)\n        img2 = img2.to(DEVICE)\n        target = target.to(DEVICE)\n        \n        optimizer.zero_grad()\n        \n        with autocast():\n            emb1, emb2 = model(img1, img2)\n            loss = criterion(emb1, emb2, target)\n        \n        scaler.scale(loss).backward()\n        scaler.step(optimizer)\n        scaler.update()\n        \n        running_loss += loss.item()\n        pbar.set_postfix(loss=f\"{loss.item():.4f}\")\n    \n    return running_loss / len(loader)\n\n# ===========================\n# VALIDATION\n# ===========================\n@torch.no_grad()\ndef validate(model, pairs_file, img_root):\n    model.eval()\n    \n    with open(pairs_file) as f:\n        pairs = [line.strip().split() for line in f if line.strip()]\n    \n    unique_paths = sorted(set(img1 for img1, _, _ in pairs) | \n                          set(img2 for _, img2, _ in pairs))\n    path_to_idx = {path: i for i, path in enumerate(unique_paths)}\n    \n    val_ds = VerificationDataset(unique_paths, img_root)\n    val_loader = DataLoader(val_ds, batch_size=128, num_workers=CFG['num_workers'], shuffle=False)\n    \n    all_embeddings = []\n    for imgs in val_loader:\n        imgs = imgs.to(DEVICE)\n        with autocast():\n            embs = model.get_embedding(imgs)\n        all_embeddings.append(embs.cpu())\n    \n    all_embeddings = torch.cat(all_embeddings, dim=0)\n    \n    similarities, labels = [], []\n    for img1, img2, label in pairs:\n        idx1, idx2 = path_to_idx[img1], path_to_idx[img2]\n        sim = F.cosine_similarity(all_embeddings[idx1].unsqueeze(0), \n                                  all_embeddings[idx2].unsqueeze(0)).item()\n        similarities.append(sim)\n        labels.append(int(label))\n    \n    auc = roc_auc_score(labels, similarities)\n    fpr, tpr, _ = roc_curve(labels, similarities)\n    eer = fpr[np.nanargmin(np.abs(tpr - (1 - fpr)))]\n    \n    return auc, eer","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ===========================\n# MAIN\n# ===========================\nif __name__ == '__main__':\n    train_ds = SiamesePairDataset(CFG['data_root'])\n    train_loader = DataLoader(train_ds, batch_size=CFG['batch_size'], \n                              num_workers=CFG['num_workers'], pin_memory=True)\n    \n    model = SiameseNetwork(CFG['emb_dim']).to(DEVICE)\n    criterion = ContrastiveLoss(margin=1.0)\n    optimizer = torch.optim.AdamW(model.parameters(), lr=CFG['lr'], weight_decay=1e-4)\n    scaler = GradScaler()\n    \n    best_auc = 0\n    for epoch in range(1, CFG['epochs'] + 1):\n        train_loss = train_one_epoch(model, train_loader, optimizer, criterion, scaler, epoch)\n        auc, eer = validate(model, CFG['ver_pairs'], CFG['ver_root'])\n        \n        print(f'Loss: {train_loss:.4f} | AUC: {auc:.4f} | EER: {eer:.2%}')\n        \n        if auc > best_auc:\n            best_auc = auc\n            torch.save(model.state_dict(), CFG['save_path'])\n            print(f'Best model saved (AUC: {auc:.4f})')\n    \n    print(f'\\nTraining Complete - Best AUC: {best_auc:.4f}')","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2025-11-18T07:17:43.187812Z","iopub.execute_input":"2025-11-18T07:17:43.188461Z","iopub.status.idle":"2025-11-18T07:39:24.618284Z","shell.execute_reply.started":"2025-11-18T07:17:43.188433Z","shell.execute_reply":"2025-11-18T07:39:24.617332Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}