{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceType":"competition","sourceId":19991,"databundleVersionId":1117522},{"sourceType":"datasetVersion","sourceId":14646030,"datasetId":9355878,"databundleVersionId":15485873},{"sourceType":"datasetVersion","sourceId":15521737,"datasetId":9930532,"databundleVersionId":16448715},{"sourceType":"datasetVersion","sourceId":15823603,"datasetId":10143131,"databundleVersionId":16772608}],"dockerImageVersionId":31260,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"# \"\"\"\n# Steganalyzer v4  —  Binary: COVER vs DCT_STEGO\n# ================================================\n# Fixes class collapse by removing LSB and going binary.\n# DCT method: quantized 8x8 blocks, LSB in non-zero AC coefficients (zigzag)\n\n# CONFIG at the bottom — set COVER_DIR and DCT_DIR.\n# \"\"\"\n\n# import os, random, warnings, time\n# from pathlib import Path\n# import numpy as np\n# from PIL import Image\n# from scipy.fftpack import dct as sdct, idct as sidct\n# import torch, torch.nn as nn, torch.nn.functional as F\n# from torch.utils.data import Dataset, DataLoader\n# from sklearn.metrics import precision_recall_fscore_support, confusion_matrix\n# try:\n#     from tqdm import tqdm\n#     HAS_TQDM = True\n# except ImportError:\n#     HAS_TQDM = False\n# warnings.filterwarnings('ignore')\n\n# CLASS_NAMES = ['COVER', 'DCT_STEGO']\n\n# ZIGZAG = np.array([\n#      0, 1, 8,16, 9, 2, 3,10,17,24,32,25,18,11, 4, 5,\n#     12,19,26,33,40,48,41,34,27,20,13, 6, 7,14,21,28,\n#     35,42,49,56,57,50,43,36,29,22,15,23,30,37,44,51,\n#     58,59,52,45,38,31,39,46,53,60,61,54,47,55,62,63])\n\n# QUANT = np.array([\n#     [16,11,10,16,24,40,51,61],[12,12,14,19,26,58,60,55],\n#     [14,13,16,24,40,57,69,56],[14,17,22,29,51,87,80,62],\n#     [18,22,37,56,68,109,103,77],[24,35,55,64,81,104,113,92],\n#     [49,64,78,87,103,121,120,101],[72,92,95,98,112,100,103,99]\n# ], dtype=np.float32)\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # EMBEDDING  (only DCT — your exact zigzag method)\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _qt(quality):\n#     s = 5000/quality if quality < 50 else 200 - 2*quality\n#     return np.clip(np.floor((QUANT*s + 50)/100), 1, 255)\n\n# def embed_dct(cover: np.ndarray, rate=0.4, quality=75) -> np.ndarray:\n#     \"\"\"Your exact method: quantize → zigzag → LSB replace in non-zero AC → reconstruct.\"\"\"\n#     h, w   = cover.shape\n#     h8, w8 = h - h%8, w - w%8\n#     img    = cover[:h8, :w8].astype(np.float32) - 128\n#     qt     = _qt(quality)\n#     hb, wb = h8//8, w8//8\n#     blocks = np.zeros((hb, wb, 8, 8), dtype=np.int16)\n#     for i in range(hb):\n#         for j in range(wb):\n#             b = img[i*8:(i+1)*8, j*8:(j+1)*8]\n#             blocks[i,j] = np.round(sdct(sdct(b.T,norm='ortho').T,norm='ortho')/qt)\n\n#     n_bits  = int(hb * wb * 63 * rate)\n#     bits    = np.random.randint(0, 2, n_bits, dtype=np.int16)\n#     bit_idx = 0\n#     stego   = blocks.copy()\n#     for i in range(hb):\n#         for j in range(wb):\n#             flat = stego[i,j].flatten()[ZIGZAG]\n#             for k in range(1, 64):\n#                 if bit_idx >= n_bits: break\n#                 if flat[k] != 0:\n#                     flat[k] = (flat[k] & ~1) | bits[bit_idx]\n#                     bit_idx += 1\n#             inv = np.empty(64, dtype=np.int16)\n#             inv[ZIGZAG] = flat\n#             stego[i,j] = inv.reshape(8,8)\n#             if bit_idx >= n_bits: break\n#         if bit_idx >= n_bits: break\n\n#     out = np.zeros((h8, w8), dtype=np.float32)\n#     for i in range(hb):\n#         for j in range(wb):\n#             b = stego[i,j].astype(np.float32) * qt\n#             out[i*8:(i+1)*8, j*8:(j+1)*8] = sidct(sidct(b.T,norm='ortho').T,norm='ortho')\n#     return np.clip(out+128, 0, 255).astype(np.uint8)\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # SRM  (30 fixed high-pass kernels)\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _srm_kernels():\n#     F=[]; z=lambda: np.zeros((5,5))\n#     f=z(); f[2,1:4]=[-1,2,-1];      F.append(f/2)\n#     f=z(); f[1:4,2]=[-1,2,-1];      F.append(f/2)\n#     f=z(); f[2,0:5]=[-1,2,-6,2,-1]; F.append(f/4)\n#     f=z(); f[:,2]  =[-1,2,-6,2,-1]; F.append(f/4)\n#     f=z(); f[1,1]=f[3,3]=-1; f[2,2]=2; F.append(f/2)\n#     f=z(); f[1,3]=f[3,1]=-1; f[2,2]=2; F.append(f/2)\n#     f=z(); f[2,1:5]=[-1,3,-3,1];    F.append(f/2)\n#     f=z(); f[1:5,2]=[-1,3,-3,1];    F.append(f/2)\n#     F.append(np.array([[0,0,-1,0,0],[0,0,2,0,0],[-1,2,-4,2,-1],[0,0,2,0,0],[0,0,-1,0,0]])/4)\n#     F.append(np.array([[0,0,0,0,0],[0,-1,2,-1,0],[0,2,-4,2,0],[0,-1,2,-1,0],[0,0,0,0,0]])/4)\n#     kb=np.array([5,5,5,-3,0,-3,-3,-3,-3],dtype=float).reshape(3,3)/15\n#     for r in range(8):\n#         f=z(); f[1:4,1:4]=np.rot90(kb,r%4)*(1 if r<4 else -1); F.append(f)\n#     for dy,dx in [(0,1),(0,-1),(1,0),(-1,0),(1,1),(-1,-1),(1,-1),(-1,1)]:\n#         f=z(); f[2,2]=1; f[2+dy,2+dx]=-1; F.append(f)\n#     while len(F)<30:\n#         f=z(); f[2,2]=1; f[2,1]=f[2,3]=f[1,2]=f[3,2]=-0.25; F.append(f)\n#     return torch.tensor(np.stack(F[:30])[:,None], dtype=torch.float32)\n\n# class SRM(nn.Module):\n#     def __init__(self):\n#         super().__init__()\n#         self.register_buffer('w', _srm_kernels())\n#     def forward(self, x):\n#         x = x.mean(dim=1, keepdim=True)          # YCbCr → Y only\n#         return torch.tanh(F.conv2d(x*255, self.w, padding=2)/3)\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # MODEL\n# # ═══════════════════════════════════════════════════════════════════\n\n# class SE(nn.Module):\n#     def __init__(self, c, r=8):\n#         super().__init__()\n#         self.fc = nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(),\n#                                 nn.Linear(c,c//r,False), nn.ReLU(True),\n#                                 nn.Linear(c//r,c,False), nn.Sigmoid())\n#     def forward(self,x): return x * self.fc(x).view(x.size(0),-1,1,1)\n\n# class Blk(nn.Module):\n#     def __init__(self, ci, co, se=False):\n#         super().__init__()\n#         self.net  = nn.Sequential(\n#             nn.Conv2d(ci,co,3,padding=1,bias=False), nn.BatchNorm2d(co), nn.ReLU(True),\n#             nn.Conv2d(co,co,3,padding=1,bias=False), nn.BatchNorm2d(co))\n#         self.skip = nn.Identity() if ci==co else nn.Conv2d(ci,co,1,bias=False)\n#         self.se   = SE(co) if se else nn.Identity()\n#     def forward(self,x): return F.relu(self.se(self.net(x))+self.skip(x),True)\n\n# def _dn(ci,co):\n#     return nn.Sequential(nn.Conv2d(ci,co,3,stride=2,padding=1,bias=False),\n#                          nn.BatchNorm2d(co), nn.ReLU(True))\n\n# class SRNet(nn.Module):\n#     def __init__(self, nc=2, drop=0.3):\n#         super().__init__()\n#         self.srm   = SRM()\n#         self.entry = nn.Sequential(nn.Conv2d(30,64,3,padding=1,bias=False),\n#                                    nn.BatchNorm2d(64), nn.ReLU(True))\n#         self.s1=nn.Sequential(Blk(64,64),Blk(64,64));               self.d1=_dn(64,64)\n#         self.s2=nn.Sequential(Blk(64,128),Blk(128,128));            self.d2=_dn(128,128)\n#         self.s3=nn.Sequential(Blk(128,256,True),Blk(256,256,True)); self.d3=_dn(256,256)\n#         self.s4=nn.Sequential(Blk(256,512,True),Blk(512,512,True))\n#         self.head=nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(),\n#                                 nn.Dropout(drop), nn.Linear(512,nc))\n#     def forward(self,x):\n#         x=self.entry(self.srm(x))\n#         x=self.d1(self.s1(x)); x=self.d2(self.s2(x))\n#         x=self.d3(self.s3(x)); return self.head(self.s4(x))\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # DATASET  —  strictly 50/50 balanced: 1 cover → 1 stego\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _paths(folder, limit):\n#     if not folder or not os.path.exists(folder): return []\n#     exts = {'.jpg','.jpeg','.png','.bmp','.pgm'}\n#     return [str(p) for p in sorted(Path(folder).rglob('*'))\n#             if p.suffix.lower() in exts][:limit]\n\n# def _chip(arr, size, train):\n#     h,w = arr.shape[0], arr.shape[1]\n#     if train and h>size and w>size:\n#         y=random.randint(0,h-size); x=random.randint(0,w-size)\n#         arr = arr[y:y+size, x:x+size] if arr.ndim==2 else arr[y:y+size, x:x+size,:]\n#         if random.random()<.5: arr=(arr[:,::-1] if arr.ndim==2 else arr[:,::-1,:]).copy()\n#         if random.random()<.5: arr=(arr[::-1]   if arr.ndim==2 else arr[::-1,:,:]).copy()\n#     else:\n#         arr=np.array(Image.fromarray(arr).resize((size,size), Image.BILINEAR))\n#     return arr\n\n# def _to_tensor(arr):\n#     t = torch.from_numpy(arr.astype(np.float32)/255).permute(2,0,1)\n#     return (t - 0.5) / 0.5\n\n# class StegoDataset(Dataset):\n#     \"\"\"\n#     Binary dataset: 0=COVER, 1=DCT_STEGO.\n#     Perfectly balanced — idx%2 determines class, idx//2 selects the image.\n#     DCT_DIR: pre-made stego (ALASKA2 JMiPOD etc). None = generate on-the-fly.\n#     \"\"\"\n#     def __init__(self, cover_dir, dct_dir=None, limit=2000, size=256, train=True):\n#         self.covers  = _paths(cover_dir, limit)\n#         self.dct_pre = _paths(dct_dir,   limit)\n#         self.size=size; self.train=train\n#         # Match ALASKA2 payload spec: 0.4 bpp avg, quality 75/90/95\n#         self.rates   = [0.2, 0.3, 0.4, 0.5]\n#         self.quals   = [75, 90, 95]\n#         assert self.covers, f\"No images found in: {cover_dir}\"\n#         n_stego = len(self.dct_pre) if self.dct_pre else len(self.covers)\n#         print(f\"  [{'TRAIN' if train else 'VAL'}] \"\n#               f\"covers={len(self.covers)} | \"\n#               f\"stego={'pre-loaded ('+str(n_stego)+')' if self.dct_pre else 'generated'} | \"\n#               f\"total samples={len(self.covers)*2:,}\")\n\n#     def __len__(self): return len(self.covers) * 2   # 50% cover, 50% stego\n\n#     def __getitem__(self, idx):\n#         lbl = idx % 2          # 0=cover, 1=stego  — perfectly alternating\n#         ii  = idx // 2\n\n#         ycbcr = np.array(Image.open(self.covers[ii]).convert('YCbCr'))\n\n#         if lbl == 0:           # COVER — no changes\n#             img = _chip(ycbcr, self.size, self.train)\n\n#         else:                  # DCT STEGO\n#             ycbcr2 = np.array(Image.open(self.dct_pre[ii]).convert('YCbCr'))\n#             img = _chip(ycbcr2, self.size, self.train)\n\n#         return _to_tensor(img), lbl\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # LOSS  —  standard BCE / cross-entropy, NO class weighting needed\n# #          (dataset is already 50/50 balanced)\n# # ═══════════════════════════════════════════════════════════════════\n\n# class LabelSmoothCE(nn.Module):\n#     def __init__(self, ls=0.05):\n#         super().__init__(); self.ls=ls\n#     def forward(self, logits, tgt):\n#         n=logits.size(1)\n#         sm=torch.full_like(logits, self.ls/(n-1))\n#         sm.scatter_(1, tgt.unsqueeze(1), 1-self.ls)\n#         return -(sm * F.log_softmax(logits,1)).sum(1).mean()\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # TRAIN / EVAL\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _epoch(model, loader, crit, opt, scaler, device, train, ep, epochs):\n#     model.train(train)\n#     ls=cor=tot=0\n#     phase = \"TRAIN\" if train else \"VAL  \"\n#     iterator = tqdm(loader, desc=f\"  Ep{ep:02d}/{epochs} {phase}\", ncols=88,\n#                     leave=False) if HAS_TQDM else loader\n\n#     ctx = torch.enable_grad if train else torch.no_grad\n#     with ctx():\n#         for bi,(x,y) in enumerate(iterator):\n#             x,y = x.to(device), y.to(device)\n#             with torch.amp.autocast('cuda', enabled=device.type=='cuda'):\n#                 out=model(x); loss=crit(out,y)\n#             if train:\n#                 opt.zero_grad(); scaler.scale(loss).backward()\n#                 scaler.unscale_(opt)\n#                 nn.utils.clip_grad_norm_(model.parameters(), 5)\n#                 scaler.step(opt); scaler.update()\n#             ls+=loss.item(); pred=out.argmax(1)\n#             tot+=y.size(0); cor+=(pred==y).sum().item()\n#             if HAS_TQDM:\n#                 iterator.set_postfix(loss=f\"{ls/(bi+1):.3f}\", acc=f\"{100*cor/tot:.1f}%\")\n#             elif bi%20==0:\n#                 print(f\"    {bi+1}/{len(loader)} loss={ls/(bi+1):.3f} acc={100*cor/max(tot,1):.1f}%\", end='\\r')\n#     if not HAS_TQDM: print()\n#     return ls/len(loader), 100*cor/tot\n\n# def _eval(model, loader, crit, device):\n#     model.eval(); ls=cor=tot=0; ps=[]; ys=[]\n#     with torch.no_grad():\n#         for x,y in loader:\n#             x,y=x.to(device),y.to(device)\n#             with torch.amp.autocast('cuda', enabled=device.type=='cuda'):\n#                 out=model(x); loss=crit(out,y)\n#             ls+=loss.item(); pred=out.argmax(1)\n#             tot+=y.size(0); cor+=(pred==y).sum().item()\n#             ps.extend(pred.cpu().tolist()); ys.extend(y.cpu().tolist())\n#     _,_,f,_ = precision_recall_fscore_support(ys,ps,average='binary',zero_division=0)\n#     cm = confusion_matrix(ys,ps,labels=[0,1])\n#     cover_acc  = 100*sum(p==l for p,l in zip(ps,ys) if l==0)/max(ys.count(0),1)\n#     stego_acc  = 100*sum(p==l for p,l in zip(ps,ys) if l==1)/max(ys.count(1),1)\n#     return ls/len(loader), 100*cor/tot, cover_acc, stego_acc, f*100, cm\n\n# def train_model(model, tr, va, device, epochs=30, lr=1e-4, wd=1e-4, save='best.pth'):\n#     crit = LabelSmoothCE(ls=0.05)\n#     opt  = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=wd)\n#     sch  = torch.optim.lr_scheduler.CosineAnnealingLR(opt, epochs, lr*0.01)\n#     sc   = torch.amp.GradScaler('cuda', enabled=device.type=='cuda')\n#     best_acc = best_stego = 0.\n\n#     print(f\"\\n{'='*60}\")\n#     print(f\"  Training {epochs} epochs | {len(tr)} batches/ep | {device}\")\n#     print(f\"{'='*60}\\n\")\n\n#     for ep in range(1, epochs+1):\n#         t0 = time.time()\n#         print(f\"┌─ Epoch {ep}/{epochs} {'─'*44}\")\n#         tr_loss, tr_acc = _epoch(model,tr,crit,opt,sc,device,True, ep,epochs)\n#         va_loss, va_acc, cov_acc, stg_acc, f1, cm = _eval(model,va,crit,device)\n#         sch.step()\n#         elapsed = time.time()-t0\n\n#         print(f\"│  Loss   : train={tr_loss:.4f}  val={va_loss:.4f}\")\n#         print(f\"│  Acc    : train={tr_acc:.1f}%  val={va_acc:.1f}%  F1={f1:.1f}%\")\n#         print(f\"│  Class  : COVER={cov_acc:.1f}%  DCT_STEGO={stg_acc:.1f}%\")\n#         print(f\"│  LR     : {sch.get_last_lr()[0]:.2e}  |  time: {elapsed:.0f}s\")\n\n#         saved=[]\n#         if va_acc > best_acc:\n#             best_acc = va_acc\n#             torch.save({'ep':ep,'state':model.state_dict(),'acc':va_acc,'cm':cm.tolist()}, save)\n#             saved.append(f\"best overall ({va_acc:.2f}%)\")\n#         if stg_acc > best_stego:\n#             best_stego = stg_acc\n#             torch.save(model.state_dict(), save.replace('.pth','_bestStego.pth'))\n#             saved.append(f\"best stego ({stg_acc:.2f}%)\")\n#         if saved: print(f\"│  ✓ Saved: {' | '.join(saved)}\")\n#         print(f\"└─{'─'*53}\\n\")\n\n#     cm=np.array(torch.load(save,map_location='cpu')['cm'])\n#     print(f\"\\n{'='*60}\")\n#     print(f\"  CONFUSION MATRIX (rows=true, cols=predicted)\")\n#     print(f\"{'':>14}{'COVER':>14}{'DCT_STEGO':>14}\")\n#     for i,n in enumerate(CLASS_NAMES):\n#         print(f\"{n:>14}\" + \"\".join(f\"{cm[i,j]:>14}\" for j in range(2)))\n#     print(f\"\\n  Best val={best_acc:.2f}%  |  Best stego={best_stego:.2f}%\")\n#     print(f\"{'='*60}\\n\")\n#     return best_acc\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # DETECT\n# # ═══════════════════════════════════════════════════════════════════\n\n# @torch.no_grad()\n# def detect(model, image, device, size=256):\n#     model.eval()\n#     if isinstance(image,str): pil=Image.open(image)\n#     elif isinstance(image,np.ndarray): pil=Image.fromarray(image)\n#     else: pil=image\n#     arr=np.array(pil.convert('YCbCr').resize((size,size), Image.BILINEAR))\n#     t=((torch.from_numpy(arr.astype(np.float32)/255).permute(2,0,1)-0.5)/0.5)\n#     probs=torch.softmax(model(t.unsqueeze(0).to(device)),1)[0].cpu().tolist()\n#     pred=int(np.argmax(probs))\n#     print(f\"\\nPrediction : {CLASS_NAMES[pred]}  ({max(probs)*100:.1f}% confidence)\")\n#     for i,n in enumerate(CLASS_NAMES):\n#         print(f\"  {n:<14} {probs[i]*100:5.1f}%  {'█'*int(probs[i]*30)}\")\n#     return pred, probs\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # SEED + MAIN\n# # ═══════════════════════════════════════════════════════════════════\n\n# def set_seed(s=42):\n#     random.seed(s); np.random.seed(s)\n#     torch.manual_seed(s); torch.cuda.manual_seed_all(s)\n\n# if __name__ == '__main__':\n\n#     # ╔════════════════════════════════════════╗\n#     # ║            CONFIG                      ║\n#     # ╠════════════════════════════════════════╣\n#     COVER_DIR  = '/kaggle/input/datasets/mdezazulhaquemahi/analysis/steganalysis/Train'\n#     DCT_DIR    = '/kaggle/input/datasets/mdezazulhaquemahi/analysis/steganalysis/Dct'\n#     TEST_IMAGE = '/kaggle/input/testpic/bLVquIQ.jpg'  # optional test\n#     NUM_TRAIN  = 1000    # adjust based on your dataset size\n#     NUM_VAL    = 200\n#     IMAGE_SIZE = 256\n#     BATCH_SIZE = 16\n#     EPOCHS     = 30\n#     LR = 0.001        # learning rate\n#     SAVE       = 'steganalyzer_v4.pth'\n#     # ╚════════════════════════════════════════╝\n\n#     set_seed(42)\n#     DEV = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n\n#     print(f\"\\n{'='*60}\")\n#     print(f\"  STEGANALYZER v4  —  {' / '.join(CLASS_NAMES)}  (binary)\")\n#     print(f\"{'='*60}\")\n#     print(f\"  Device : {DEV}\" + (f\" ({torch.cuda.get_device_name(0)})\" if DEV.type=='cuda' else \"\"))\n#     print(f\"  Size   : {IMAGE_SIZE}×{IMAGE_SIZE}  |  Batch: {BATCH_SIZE}  |  Epochs: {EPOCHS}\")\n#     print(f\"  tqdm   : {'yes' if HAS_TQDM else 'no — pip install tqdm'}\")\n#     print(f\"\\n  Loading datasets...\")\n\n#     tr_ds = StegoDataset(COVER_DIR, DCT_DIR, NUM_TRAIN, IMAGE_SIZE, train=True)\n#     va_ds = StegoDataset(COVER_DIR, DCT_DIR, NUM_VAL,   IMAGE_SIZE, train=False)\n#     print(f\"  Batches : train={len(tr_ds)//BATCH_SIZE}  val={len(va_ds)//BATCH_SIZE}\")\n\n#     tr_ld = DataLoader(tr_ds, BATCH_SIZE, shuffle=True,  num_workers=2, pin_memory=True)\n#     va_ld = DataLoader(va_ds, BATCH_SIZE, shuffle=False, num_workers=2, pin_memory=True)\n\n#     model  = SRNet(nc=2, drop=0.3).to(DEV)\n#     params = sum(p.numel() for p in model.parameters())\n#     print(f\"  Model  : SRNet  ({params:,} params)\\n\")\n\n#     train_model(model, tr_ld, va_ld, DEV, EPOCHS, LR, save=SAVE)\n\n#     ckpt = torch.load(SAVE, map_location=DEV)\n#     model.load_state_dict(ckpt['state'])\n#     if os.path.exists(TEST_IMAGE):\n#         detect(model, TEST_IMAGE, DEV, IMAGE_SIZE)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-19T14:16:07.494509Z","iopub.execute_input":"2026-04-19T14:16:07.495266Z","iopub.status.idle":"2026-04-19T14:23:25.474546Z","shell.execute_reply.started":"2026-04-19T14:16:07.495225Z","shell.execute_reply":"2026-04-19T14:23:25.473453Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# \"\"\"\n# Steganalyzer v6  —  Binary: COVER vs DCT_STEGO\n# ================================================\n# Changes from v5:\n#   [CHANGED] Loss: FocalSmoothCE → nn.CrossEntropyLoss()\n#   [CHANGED] DataLoader: removed WeightedRandomSampler → standard shuffle=True\n#   [CHANGED] Dropout: 0.4 → 0.25\n#   [CHANGED] Dataset: on-the-fly DCT embedding (embed_dct) as primary source,\n#             pre-made DCT_DIR used as fallback when USE_PREMADE=True\n#   [UNCHANGED] SRNet architecture, SRM filters, LR=1e-4, cosine+warmup,\n#               gradient clipping, mixed precision, collapse guard,\n#               YCbCr preprocessing, per-class acc + F1 printing\n# \"\"\"\n\n# import os, random, warnings, time\n# from pathlib import Path\n# import numpy as np\n# from PIL import Image\n# from scipy.fftpack import dct as sdct, idct as sidct\n# import torch, torch.nn as nn, torch.nn.functional as F\n# from torch.utils.data import Dataset, DataLoader\n# from sklearn.metrics import precision_recall_fscore_support, confusion_matrix\n# try:\n#     from tqdm import tqdm\n#     HAS_TQDM = True\n# except ImportError:\n#     HAS_TQDM = False\n# warnings.filterwarnings('ignore')\n\n# CLASS_NAMES = ['COVER', 'DCT_STEGO']\n\n# ZIGZAG = np.array([\n#      0, 1, 8,16, 9, 2, 3,10,17,24,32,25,18,11, 4, 5,\n#     12,19,26,33,40,48,41,34,27,20,13, 6, 7,14,21,28,\n#     35,42,49,56,57,50,43,36,29,22,15,23,30,37,44,51,\n#     58,59,52,45,38,31,39,46,53,60,61,54,47,55,62,63])\n\n# QUANT = np.array([\n#     [16,11,10,16,24,40,51,61],[12,12,14,19,26,58,60,55],\n#     [14,13,16,24,40,57,69,56],[14,17,22,29,51,87,80,62],\n#     [18,22,37,56,68,109,103,77],[24,35,55,64,81,104,113,92],\n#     [49,64,78,87,103,121,120,101],[72,92,95,98,112,100,103,99]\n# ], dtype=np.float32)\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # [UNCHANGED] DCT EMBEDDING — your exact method from embedding script\n# # ═══════════════════════════════════════════════════════════════════\n\n# def embed_dct(cover_y: np.ndarray, rate=0.3) -> np.ndarray:\n#     \"\"\"\n#     Matches your embedding script exactly:\n#       Y-channel only, rate=0.3, QUANT table, zigzag LSB in non-zero AC.\n#     Input/output: uint8 grayscale (Y channel).\n#     \"\"\"\n#     h, w   = cover_y.shape\n#     h8, w8 = h - h%8, w - w%8\n#     img    = cover_y[:h8, :w8].astype(np.float32) - 128\n#     hb, wb = h8//8, w8//8\n\n#     total_coeffs = hb * wb * 63\n#     num_bits     = int(total_coeffs * rate)\n#     bits         = np.random.randint(0, 2, num_bits, dtype=np.int16)\n#     bit_idx      = 0\n\n#     blocks = np.zeros((hb, wb, 8, 8), dtype=np.int16)\n#     for i in range(hb):\n#         for j in range(wb):\n#             b = img[i*8:(i+1)*8, j*8:(j+1)*8]\n#             blocks[i,j] = np.round(\n#                 sdct(sdct(b.T, norm='ortho').T, norm='ortho') / QUANT)\n\n#     stego = blocks.copy()\n#     for i in range(hb):\n#         for j in range(wb):\n#             flat = stego[i,j].flatten()[ZIGZAG]\n#             for k in range(1, 64):\n#                 if bit_idx >= num_bits: break\n#                 if flat[k] != 0:\n#                     flat[k] = (flat[k] & ~1) | bits[bit_idx]\n#                     bit_idx += 1\n#             inv         = np.empty(64, dtype=np.int16)\n#             inv[ZIGZAG] = flat\n#             stego[i,j]  = inv.reshape(8, 8)\n#             if bit_idx >= num_bits: break\n#         if bit_idx >= num_bits: break\n\n#     out = np.zeros((h8, w8), dtype=np.float32)\n#     for i in range(hb):\n#         for j in range(wb):\n#             b = stego[i,j].astype(np.float32) * QUANT\n#             out[i*8:(i+1)*8, j*8:(j+1)*8] = \\\n#                 sidct(sidct(b.T, norm='ortho').T, norm='ortho')\n\n#     result = cover_y.copy()\n#     result[:h8, :w8] = np.clip(out + 128, 0, 255).astype(np.uint8)\n#     return result\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # [UNCHANGED] SRM filters\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _srm_kernels():\n#     F=[]; z=lambda: np.zeros((5,5))\n#     f=z(); f[2,1:4]=[-1,2,-1];      F.append(f/2)\n#     f=z(); f[1:4,2]=[-1,2,-1];      F.append(f/2)\n#     f=z(); f[2,0:5]=[-1,2,-6,2,-1]; F.append(f/4)\n#     f=z(); f[:,2]  =[-1,2,-6,2,-1]; F.append(f/4)\n#     f=z(); f[1,1]=f[3,3]=-1; f[2,2]=2; F.append(f/2)\n#     f=z(); f[1,3]=f[3,1]=-1; f[2,2]=2; F.append(f/2)\n#     f=z(); f[2,1:5]=[-1,3,-3,1];    F.append(f/2)\n#     f=z(); f[1:5,2]=[-1,3,-3,1];    F.append(f/2)\n#     F.append(np.array([[0,0,-1,0,0],[0,0,2,0,0],[-1,2,-4,2,-1],[0,0,2,0,0],[0,0,-1,0,0]])/4)\n#     F.append(np.array([[0,0,0,0,0],[0,-1,2,-1,0],[0,2,-4,2,0],[0,-1,2,-1,0],[0,0,0,0,0]])/4)\n#     kb=np.array([5,5,5,-3,0,-3,-3,-3,-3],dtype=float).reshape(3,3)/15\n#     for r in range(8):\n#         f=z(); f[1:4,1:4]=np.rot90(kb,r%4)*(1 if r<4 else -1); F.append(f)\n#     for dy,dx in [(0,1),(0,-1),(1,0),(-1,0),(1,1),(-1,-1),(1,-1),(-1,1)]:\n#         f=z(); f[2,2]=1; f[2+dy,2+dx]=-1; F.append(f)\n#     while len(F)<30:\n#         f=z(); f[2,2]=1; f[2,1]=f[2,3]=f[1,2]=f[3,2]=-0.25; F.append(f)\n#     return torch.tensor(np.stack(F[:30])[:,None], dtype=torch.float32)\n\n# class SRM(nn.Module):\n#     def __init__(self):\n#         super().__init__()\n#         self.register_buffer('w', _srm_kernels())\n#     def forward(self, x):\n#         x = x.mean(dim=1, keepdim=True)\n#         return torch.tanh(F.conv2d(x*255, self.w, padding=2)/3)\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # [UNCHANGED] SRNet architecture  |  [CHANGED] drop default: 0.4→0.25\n# # ═══════════════════════════════════════════════════════════════════\n\n# class SE(nn.Module):\n#     def __init__(self, c, r=8):\n#         super().__init__()\n#         self.fc = nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(),\n#                                 nn.Linear(c,c//r,False), nn.ReLU(True),\n#                                 nn.Linear(c//r,c,False), nn.Sigmoid())\n#     def forward(self,x): return x * self.fc(x).view(x.size(0),-1,1,1)\n\n# class Blk(nn.Module):\n#     def __init__(self, ci, co, se=False):\n#         super().__init__()\n#         self.net  = nn.Sequential(\n#             nn.Conv2d(ci,co,3,padding=1,bias=False), nn.BatchNorm2d(co), nn.ReLU(True),\n#             nn.Conv2d(co,co,3,padding=1,bias=False), nn.BatchNorm2d(co))\n#         self.skip = nn.Identity() if ci==co else nn.Conv2d(ci,co,1,bias=False)\n#         self.se   = SE(co) if se else nn.Identity()\n#     def forward(self,x): return F.relu(self.se(self.net(x))+self.skip(x),True)\n\n# def _dn(ci,co):\n#     return nn.Sequential(nn.Conv2d(ci,co,3,stride=2,padding=1,bias=False),\n#                          nn.BatchNorm2d(co), nn.ReLU(True))\n\n# class SRNet(nn.Module):\n#     def __init__(self, nc=2, drop=0.25):   # [CHANGED] drop 0.4 → 0.25\n#         super().__init__()\n#         self.srm   = SRM()\n#         self.entry = nn.Sequential(nn.Conv2d(30,64,3,padding=1,bias=False),\n#                                    nn.BatchNorm2d(64), nn.ReLU(True))\n#         self.s1=nn.Sequential(Blk(64,64),Blk(64,64));               self.d1=_dn(64,64)\n#         self.s2=nn.Sequential(Blk(64,128),Blk(128,128));            self.d2=_dn(128,128)\n#         self.s3=nn.Sequential(Blk(128,256,True),Blk(256,256,True)); self.d3=_dn(256,256)\n#         self.s4=nn.Sequential(Blk(256,512,True),Blk(512,512,True))\n#         self.head=nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(),\n#                                 nn.Dropout(drop), nn.Linear(512,nc))\n#     def forward(self,x):\n#         x=self.entry(self.srm(x))\n#         x=self.d1(self.s1(x)); x=self.d2(self.s2(x))\n#         x=self.d3(self.s3(x)); return self.head(self.s4(x))\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # DATASET\n# # [CHANGED] on-the-fly DCT embedding as primary; pre-made as fallback\n# # [UNCHANGED] 50/50 via idx%2, YCbCr input, non-overlapping split\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _paths(folder, limit=None):\n#     if not folder or not os.path.exists(folder): return []\n#     exts  = {'.jpg','.jpeg','.png','.bmp','.pgm'}\n#     paths = sorted([str(p) for p in Path(folder).rglob('*')\n#                     if p.suffix.lower() in exts])\n#     return paths[:limit] if limit else paths\n\n# def split_paths(cover_dir, dct_dir, n_train, n_val):\n#     \"\"\"Non-overlapping train/val split, matched by filename stem.\"\"\"\n#     covers    = _paths(cover_dir)\n#     cover_map = {Path(p).stem: p for p in covers}\n\n#     if dct_dir and os.path.exists(dct_dir):\n#         dcts    = _paths(dct_dir)\n#         dct_map = {Path(p).stem: p for p in dcts}\n#         common  = sorted(set(cover_map.keys()) & set(dct_map.keys()))\n#     else:\n#         common  = sorted(cover_map.keys())\n#         dct_map = {}\n\n#     random.shuffle(common)\n#     if len(common) < n_train + n_val:\n#         n_train = int(len(common) * 0.85)\n#         n_val   = len(common) - n_train\n#         print(f\"  ⚠ Only {len(common)} pairs → {n_train} train / {n_val} val\")\n\n#     def _pair(stems):\n#         cvs = [cover_map[s] for s in stems]\n#         dts = [dct_map.get(s) for s in stems]   # None if no pre-made\n#         return cvs, dts\n\n#     tr_cov, tr_dct = _pair(common[:n_train])\n#     va_cov, va_dct = _pair(common[n_train:n_train+n_val])\n#     return tr_cov, tr_dct, va_cov, va_dct\n\n# def _chip(arr, size, train):\n#     h, w = arr.shape[0], arr.shape[1]\n#     if train and h >= size and w >= size:\n#         y = random.randint(0, h-size)\n#         x = random.randint(0, w-size)\n#         arr = arr[y:y+size, x:x+size] if arr.ndim==2 else arr[y:y+size, x:x+size,:]\n#         if random.random() < .5:\n#             arr = (arr[:,::-1] if arr.ndim==2 else arr[:,::-1,:]).copy()\n#         if random.random() < .5:\n#             arr = (arr[::-1]   if arr.ndim==2 else arr[::-1,:,:]).copy()\n#     else:\n#         h0, w0 = (h-size)//2, (w-size)//2\n#         if h >= size and w >= size:\n#             arr = arr[h0:h0+size, w0:w0+size] if arr.ndim==2 \\\n#                   else arr[h0:h0+size, w0:w0+size,:]\n#         else:\n#             arr = np.array(Image.fromarray(arr).resize((size,size), Image.BILINEAR))\n#     return arr\n\n# def _to_tensor(arr):   # [UNCHANGED]\n#     t = torch.from_numpy(arr.astype(np.float32)/255).permute(2,0,1)\n#     return (t - 0.5) / 0.5\n\n# class StegoDataset(Dataset):\n#     \"\"\"\n#     Binary: 0=COVER, 1=DCT_STEGO.  idx%2→label, idx//2→image.\n\n#     [CHANGED] Stego source:\n#       USE_PREMADE=True  → load from dct_paths (your pre-generated files)\n#       USE_PREMADE=False → generate on-the-fly via embed_dct() every epoch\n#                           (more variety, slower, but avoids any domain gap)\n#     Dataset is exactly 50/50 by construction — no sampler needed.\n#     \"\"\"\n#     def __init__(self, cover_paths, dct_paths, size=256, train=True,\n#                  name=\"DATASET\", use_premade=True):\n#         self.covers      = cover_paths\n#         self.dcts        = dct_paths\n#         self.size        = size\n#         self.train       = train\n#         self.use_premade = use_premade and any(p is not None for p in dct_paths)\n#         src = \"pre-made DCT_DIR\" if self.use_premade else \"on-the-fly embed_dct()\"\n#         print(f\"  [{name}] covers={len(cover_paths)} | stego={src} \"\n#               f\"| total samples={len(cover_paths)*2:,}\")\n\n#     def __len__(self): return len(self.covers) * 2\n\n#     def __getitem__(self, idx):\n#         lbl = idx % 2\n#         ii  = idx // 2\n\n#         if lbl == 0:\n#             # COVER — load as-is\n#             ycbcr = np.array(Image.open(self.covers[ii]).convert('YCbCr'))\n#             img   = _chip(ycbcr, self.size, self.train)\n\n#         else:\n#             # STEGO\n#             if self.use_premade and self.dcts[ii] is not None:\n#                 # [CHANGED option A] pre-generated file from DCT_DIR\n#                 ycbcr = np.array(Image.open(self.dcts[ii]).convert('YCbCr'))\n#             else:\n#                 # [CHANGED option B] on-the-fly: embed into same cover\n#                 ycbcr     = np.array(Image.open(self.covers[ii]).convert('YCbCr'))\n#                 Y         = ycbcr[:,:,0]\n#                 Y_stego   = embed_dct(Y, rate=0.3)\n#                 ycbcr     = np.stack([Y_stego, ycbcr[:,:,1], ycbcr[:,:,2]], axis=2)\n#             img = _chip(ycbcr, self.size, self.train)\n\n#         return _to_tensor(img), lbl\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # TRAIN / EVAL\n# # [CHANGED] loss = nn.CrossEntropyLoss()\n# # [UNCHANGED] grad clip, AMP, collapse guard, per-class acc, F1\n# # ═══════════════════════════════════════════════════════════════════\n\n# def _epoch(model, loader, crit, opt, scaler, device, train, ep, epochs):\n#     model.train(train)\n#     ls=cor=tot=0\n#     phase    = \"TRAIN\" if train else \"VAL  \"\n#     iterator = (tqdm(loader, desc=f\"  Ep{ep:02d}/{epochs} {phase}\",\n#                      ncols=90, leave=False) if HAS_TQDM else loader)\n#     ctx = torch.enable_grad if train else torch.no_grad\n#     with ctx():\n#         for bi,(x,y) in enumerate(iterator):\n#             x,y = x.to(device), y.to(device)\n#             with torch.amp.autocast('cuda', enabled=device.type=='cuda'):\n#                 out  = model(x)\n#                 loss = crit(out, y)\n#             if train:\n#                 opt.zero_grad()\n#                 scaler.scale(loss).backward()\n#                 scaler.unscale_(opt)\n#                 nn.utils.clip_grad_norm_(model.parameters(), 3)  # [UNCHANGED]\n#                 scaler.step(opt); scaler.update()\n#             ls  += loss.item(); pred = out.argmax(1)\n#             tot += y.size(0);   cor += (pred==y).sum().item()\n#             if HAS_TQDM:\n#                 iterator.set_postfix(loss=f\"{ls/(bi+1):.3f}\", acc=f\"{100*cor/tot:.1f}%\")\n#     return ls/len(loader), 100*cor/tot\n\n# def _eval(model, loader, crit, device):\n#     model.eval(); ls=cor=tot=0; ps=[]; ys=[]\n#     with torch.no_grad():\n#         for x,y in loader:\n#             x,y=x.to(device),y.to(device)\n#             with torch.amp.autocast('cuda', enabled=device.type=='cuda'):\n#                 out=model(x); loss=crit(out,y)\n#             ls+=loss.item(); pred=out.argmax(1)\n#             tot+=y.size(0); cor+=(pred==y).sum().item()\n#             ps.extend(pred.cpu().tolist()); ys.extend(y.cpu().tolist())\n#     _,_,f,_ = precision_recall_fscore_support(ys,ps,average='binary',zero_division=0)\n#     cm        = confusion_matrix(ys,ps,labels=[0,1])\n#     cover_acc = 100*sum(p==l for p,l in zip(ps,ys) if l==0)/max(ys.count(0),1)\n#     stego_acc = 100*sum(p==l for p,l in zip(ps,ys) if l==1)/max(ys.count(1),1)\n#     return ls/len(loader), 100*cor/tot, cover_acc, stego_acc, f*100, cm\n\n\n# def train_model(model, tr_ld, va_ld, device, epochs=40, lr=1e-4, wd=2e-4,\n#                 warmup=5, save='best.pth'):\n\n#     # [CHANGED] Plain CrossEntropyLoss — no smoothing, no focal\n#     crit = nn.CrossEntropyLoss()\n#     # If collapse still occurs after epoch 5, uncomment this instead:\n#     # crit = nn.CrossEntropyLoss(weight=torch.tensor([1.0, 1.2]).to(device))\n\n#     opt = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=wd)\n\n#     # [UNCHANGED] cosine + linear warmup\n#     def lr_lambda(ep):\n#         if ep < warmup: return (ep+1)/warmup\n#         progress = (ep-warmup)/(epochs-warmup)\n#         return 0.01 + 0.99*0.5*(1+np.cos(np.pi*progress))\n#     sch = torch.optim.lr_scheduler.LambdaLR(opt, lr_lambda)\n#     sc  = torch.amp.GradScaler('cuda', enabled=device.type=='cuda')\n\n#     best_acc = best_stego = 0.\n#     collapse_count = 0\n\n#     print(f\"\\n{'='*62}\")\n#     print(f\"  Training {epochs} epochs | {len(tr_ld)} batches/ep | {device}\")\n#     print(f\"  LR={lr:.0e}  WD={wd:.0e}  Warmup={warmup}ep  CrossEntropyLoss\")\n#     print(f\"{'='*62}\\n\")\n\n#     for ep in range(1, epochs+1):\n#         t0 = time.time()\n#         print(f\"┌─ Epoch {ep}/{epochs} {'─'*46}\")\n\n#         tr_loss, tr_acc = _epoch(model,tr_ld,crit,opt,sc,device,True,ep,epochs)\n#         va_loss, va_acc, cov_acc, stg_acc, f1, cm = _eval(model,va_ld,crit,device)\n#         sch.step()\n#         elapsed = time.time()-t0\n\n#         # [UNCHANGED] debug printing\n#         print(f\"│  Loss   : train={tr_loss:.4f}  val={va_loss:.4f}\")\n#         print(f\"│  Acc    : train={tr_acc:.1f}%  val={va_acc:.1f}%  F1={f1:.1f}%\")\n#         print(f\"│  Class  : COVER={cov_acc:.1f}%  DCT_STEGO={stg_acc:.1f}%\")\n#         print(f\"│  LR     : {sch.get_last_lr()[0]:.2e}  |  time: {elapsed:.0f}s\")\n\n#         # [UNCHANGED] collapse guard\n#         if stg_acc < 15.0:\n#             collapse_count += 1\n#             print(f\"│  ⚠ Collapse warning ({collapse_count}/3)  stego_acc={stg_acc:.1f}%\")\n#             if collapse_count >= 3:\n#                 print(f\"│  🔄 LR reset to {lr:.0e} to escape collapse\")\n#                 for g in opt.param_groups: g['lr'] = lr\n#                 collapse_count = 0\n#         else:\n#             collapse_count = 0\n\n#         saved=[]\n#         if va_acc > best_acc and stg_acc > 20.0:\n#             best_acc = va_acc\n#             torch.save({'ep':ep,'state':model.state_dict(),\n#                         'acc':va_acc,'stego_acc':stg_acc,'cm':cm.tolist()}, save)\n#             saved.append(f\"best overall ({va_acc:.2f}%)\")\n#         if stg_acc > best_stego:\n#             best_stego = stg_acc\n#             torch.save({'ep':ep,'state':model.state_dict(),\n#                         'acc':va_acc,'stego_acc':stg_acc,'cm':cm.tolist()},\n#                        save.replace('.pth','_bestStego.pth'))\n#             saved.append(f\"best stego ({stg_acc:.2f}%)\")\n#         if saved: print(f\"│  ✓ Saved: {' | '.join(saved)}\")\n#         print(f\"└─{'─'*55}\\n\")\n\n#     best_path = save.replace('.pth','_bestStego.pth')\n#     ckpt_path = best_path if os.path.exists(best_path) else save\n#     if os.path.exists(ckpt_path):\n#         cm = np.array(torch.load(ckpt_path, map_location='cpu')['cm'])\n#         print(f\"\\n{'='*62}\")\n#         print(f\"  CONFUSION MATRIX (rows=true, cols=predicted)\")\n#         print(f\"{'':>14}{'COVER':>14}{'DCT_STEGO':>14}\")\n#         for i,n in enumerate(CLASS_NAMES):\n#             print(f\"{n:>14}\" + \"\".join(f\"{cm[i,j]:>14}\" for j in range(2)))\n#     print(f\"\\n  Best val acc={best_acc:.2f}%  |  Best stego acc={best_stego:.2f}%\")\n#     print(f\"{'='*62}\\n\")\n#     return best_acc\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # DETECT\n# # ═══════════════════════════════════════════════════════════════════\n\n# @torch.no_grad()\n# def detect(model, image, device, size=256):\n#     model.eval()\n#     if isinstance(image, str):          pil = Image.open(image)\n#     elif isinstance(image, np.ndarray): pil = Image.fromarray(image)\n#     else:                               pil = image\n#     arr   = np.array(pil.convert('YCbCr').resize((size,size), Image.BILINEAR))\n#     t     = ((torch.from_numpy(arr.astype(np.float32)/255).permute(2,0,1)-0.5)/0.5)\n#     probs = torch.softmax(model(t.unsqueeze(0).to(device)),1)[0].cpu().tolist()\n#     pred  = int(np.argmax(probs))\n#     print(f\"\\n{'─'*40}\")\n#     print(f\"  Prediction : {CLASS_NAMES[pred]}  ({max(probs)*100:.1f}% confidence)\")\n#     print(f\"{'─'*40}\")\n#     for i,n in enumerate(CLASS_NAMES):\n#         print(f\"  {n:<14} {probs[i]*100:5.1f}%  {'█'*int(probs[i]*30)}\")\n#     print(f\"{'─'*40}\\n\")\n#     return CLASS_NAMES[pred], {n:p for n,p in zip(CLASS_NAMES,probs)}\n\n\n# # ═══════════════════════════════════════════════════════════════════\n# # SEED + MAIN\n# # ═══════════════════════════════════════════════════════════════════\n\n# def set_seed(s=42):\n#     random.seed(s); np.random.seed(s)\n#     torch.manual_seed(s); torch.cuda.manual_seed_all(s)\n\n# if __name__ == '__main__':\n\n#     # ╔══════════════════════════════════════════════╗\n#     # ║                  CONFIG                      ║\n#     # ╠══════════════════════════════════════════════╣\n#     COVER_DIR   = '/kaggle/input/datasets/mdezazulhaquemahi/analysis/steganalysis/Train'\n#     DCT_DIR     = '/kaggle/input/datasets/mdezazulhaquemahi/analysis/steganalysis/Dct'\n#     TEST_IMAGE  = None\n#     NUM_TRAIN   = 5500           # ~80% of your 7k pairs\n#     NUM_VAL     = 1000           # ~15% of your 7k pairs\n#     IMAGE_SIZE  = 256\n#     BATCH_SIZE  = 16\n#     EPOCHS      = 40\n#     LR          = 1e-4           # [UNCHANGED]\n#     WEIGHT_DEC  = 2e-4\n#     WARMUP_EP   = 5\n#     USE_PREMADE = True           # True  → use pre-generated DCT_DIR (recommended)\n#                                  # False → on-the-fly embed_dct() every epoch\n#     SAVE        = 'steganalyzer_v6.pth'\n#     # ╚══════════════════════════════════════════════╝\n\n#     set_seed(42)\n#     DEV = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n\n#     print(f\"\\n{'='*62}\")\n#     print(f\"  STEGANALYZER v6  —  {' / '.join(CLASS_NAMES)}  (binary)\")\n#     print(f\"{'='*62}\")\n#     print(f\"  Device : {DEV}\" + (f\" ({torch.cuda.get_device_name(0)})\" if DEV.type=='cuda' else \"\"))\n#     print(f\"  Size   : {IMAGE_SIZE}×{IMAGE_SIZE}  |  Batch: {BATCH_SIZE}  |  Epochs: {EPOCHS}\")\n#     print(f\"  tqdm   : {'yes' if HAS_TQDM else 'no — pip install tqdm'}\")\n#     print(f\"\\n  Splitting dataset (no train/val overlap)...\")\n\n#     tr_covers, tr_dcts, va_covers, va_dcts = split_paths(\n#         COVER_DIR, DCT_DIR, NUM_TRAIN, NUM_VAL)\n\n#     print(f\"  Loading datasets...\")\n#     tr_ds = StegoDataset(tr_covers, tr_dcts, IMAGE_SIZE, train=True,\n#                          name=\"TRAIN\", use_premade=USE_PREMADE)\n#     va_ds = StegoDataset(va_covers, va_dcts, IMAGE_SIZE, train=False,\n#                          name=\"VAL  \", use_premade=USE_PREMADE)\n#     print(f\"  Batches : train={len(tr_ds)//BATCH_SIZE}  val={len(va_ds)//BATCH_SIZE}\")\n\n#     # [CHANGED] standard DataLoader — no WeightedRandomSampler\n#     # 50/50 balance is guaranteed by idx%2 label alternation\n#     tr_ld = DataLoader(tr_ds, BATCH_SIZE, shuffle=True,\n#                        num_workers=2, pin_memory=True)\n#     va_ld = DataLoader(va_ds, BATCH_SIZE, shuffle=False,\n#                        num_workers=2, pin_memory=True)\n\n#     model  = SRNet(nc=2, drop=0.25).to(DEV)   # [CHANGED] drop=0.25\n#     params = sum(p.numel() for p in model.parameters())\n#     print(f\"  Model  : SRNet  ({params:,} params)\\n\")\n\n#     train_model(model, tr_ld, va_ld, DEV,\n#                 epochs=EPOCHS, lr=LR, wd=WEIGHT_DEC,\n#                 warmup=WARMUP_EP, save=SAVE)\n\n#     best_path = SAVE.replace('.pth','_bestStego.pth')\n#     ckpt_path = best_path if os.path.exists(best_path) else SAVE\n#     if os.path.exists(ckpt_path):\n#         ckpt  = torch.load(ckpt_path, map_location=DEV)\n#         state = ckpt['state'] if 'state' in ckpt else ckpt\n#         model.load_state_dict(state)\n#         print(f\"  Loaded: {ckpt_path}\")\n\n#     if TEST_IMAGE and os.path.exists(TEST_IMAGE):\n#         detect(model, TEST_IMAGE, DEV, IMAGE_SIZE)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-19T14:23:33.175476Z","iopub.execute_input":"2026-04-19T14:23:33.175968Z","iopub.status.idle":"2026-04-19T15:20:43.629998Z","shell.execute_reply.started":"2026-04-19T14:23:33.175933Z","shell.execute_reply":"2026-04-19T15:20:43.628782Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"\"\"\"\nSteganalyzer v8  —  Binary: COVER vs DCT_STEGO\n================================================\nFixes from v7:\n  [FIX A] EPOCHS raised 10 → 50  (warmup stays 5, so 45 real learning epochs)\n           Early stopping still guards against over-running.\n  [FIX B] Vectorized embed_dct():  pure-NumPy block processing replaces the\n           three nested Python loops (DCT, zigzag LSB, IDCT).\n           Same mathematical result, ~20–40× faster on CPU.\n  [FIX C] Gaussian noise σ raised 1.0 → 3.0 and probability 0.30 → 0.50\n           so the augmentation is actually perceptible and regularising.\n           Also added random JPEG recompression (quality 70–95, 40% chance)\n           as a realistic real-world distortion.\n  [FIX D] Collapse-guard LR reset now restores the *scheduler's* current LR\n           rather than the initial LR, avoiding a sudden jump mid-annealing.\n  [FIX E] Separate held-out TEST split (10 % of data) evaluated once at the\n           very end so final numbers are not contaminated by val-set selection.\n  [UNCHANGED] SRNet architecture, SRM filters, cosine+warmup schedule,\n              gradient clipping, mixed precision, CrossEntropyLoss,\n              YCbCr preprocessing, non-overlapping split logic, all hyperparams\n              not explicitly listed above (LR, WD, dropout, batch size …).\n\"\"\"\n\nimport io, os, random, warnings, time\nfrom pathlib import Path\nimport numpy as np\nfrom PIL import Image\nfrom scipy.fftpack import dct as sdct, idct as sidct\nimport torch, torch.nn as nn, torch.nn.functional as F\nfrom torch.utils.data import Dataset, DataLoader\nfrom sklearn.metrics import precision_recall_fscore_support, confusion_matrix\ntry:\n    from tqdm import tqdm\n    HAS_TQDM = True\nexcept ImportError:\n    HAS_TQDM = False\nwarnings.filterwarnings('ignore')\n\nCLASS_NAMES = ['COVER', 'DCT_STEGO']\n\nZIGZAG = np.array([\n     0, 1, 8,16, 9, 2, 3,10,17,24,32,25,18,11, 4, 5,\n    12,19,26,33,40,48,41,34,27,20,13, 6, 7,14,21,28,\n    35,42,49,56,57,50,43,36,29,22,15,23,30,37,44,51,\n    58,59,52,45,38,31,39,46,53,60,61,54,47,55,62,63])\n\n# Inverse zigzag: position i in the flat zigzag order → 8×8 position\nZIGZAG_INV = np.argsort(ZIGZAG)          # [FIX B] precomputed once\n\nQUANT = np.array([\n    [16,11,10,16,24,40,51,61],[12,12,14,19,26,58,60,55],\n    [14,13,16,24,40,57,69,56],[14,17,22,29,51,87,80,62],\n    [18,22,37,56,68,109,103,77],[24,35,55,64,81,104,113,92],\n    [49,64,78,87,103,121,120,101],[72,92,95,98,112,100,103,99]\n], dtype=np.float32)\n\nEMBED_RATES = [0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4]\n\n\n# ═══════════════════════════════════════════════════════════════════\n# [FIX B] VECTORIZED DCT EMBEDDING\n# ═══════════════════════════════════════════════════════════════════\n\ndef embed_dct(cover_y: np.ndarray, rate=0.3) -> np.ndarray:\n    \"\"\"\n    Y-channel DCT LSB embedding — vectorized NumPy version.\n    Mathematically identical to v7 but ~20–40× faster:\n      no Python loops over blocks for DCT or IDCT;\n      zigzag LSB replacement done with array indexing.\n    Input/output: uint8 grayscale (Y channel).\n    \"\"\"\n    h, w   = cover_y.shape\n    h8, w8 = h - h % 8, w - w % 8\n    img    = cover_y[:h8, :w8].astype(np.float32) - 128\n    hb, wb = h8 // 8, w8 // 8\n\n    # ── 1. Forward DCT on all blocks at once ─────────────────────\n    # Reshape to (hb, wb, 8, 8), apply separable DCT row-then-col\n    blocks = img.reshape(hb, 8, wb, 8).transpose(0, 2, 1, 3)   # (hb,wb,8,8)\n    blocks = sdct(sdct(blocks, norm='ortho', axis=-1), norm='ortho', axis=-2)\n    blocks = np.round(blocks / QUANT).astype(np.int16)           # quantize\n\n    # ── 2. Zigzag + LSB replacement ──────────────────────────────\n    # Flatten each block in zigzag order → (hb*wb, 64)\n    flat      = blocks.reshape(-1, 64)[:, ZIGZAG]               # zigzag order\n    ac        = flat[:, 1:]                                      # AC coeffs only\n\n    total_ac  = hb * wb * 63\n    num_bits  = int(total_ac * rate)\n    bits      = np.random.randint(0, 2, num_bits, dtype=np.int16)\n\n    # Build a mask of non-zero AC positions (row-major scan, matches v7)\n    nonzero_mask = (ac != 0).ravel()                             # (hb*wb*63,)\n    positions    = np.where(nonzero_mask)[0][:num_bits]         # first num_bits non-zero\n\n    if len(positions) > 0:\n        bits_used = bits[:len(positions)]\n        row_idx   = positions // 63\n        col_idx   = positions %  63\n        ac_ravel  = ac.ravel()\n        linear    = row_idx * 63 + col_idx\n        ac_ravel[linear] = (ac_ravel[linear] & np.int16(~1)) | bits_used\n        ac        = ac_ravel.reshape(ac.shape)\n\n    flat[:, 1:] = ac\n\n    # Inverse zigzag → (hb*wb, 8, 8)\n    unzigzag    = np.empty_like(flat)\n    unzigzag[:, ZIGZAG_INV] = flat                              # undo zigzag\n    stego       = unzigzag.reshape(hb, wb, 8, 8)\n\n    # ── 3. Inverse DCT on all blocks at once ─────────────────────\n    recon = stego.astype(np.float32) * QUANT\n    recon = sidct(sidct(recon, norm='ortho', axis=-2), norm='ortho', axis=-1)\n\n    # Reassemble into image layout\n    out   = recon.transpose(0, 2, 1, 3).reshape(h8, w8)        # (h8, w8)\n\n    result = cover_y.copy()\n    result[:h8, :w8] = np.clip(out + 128, 0, 255).astype(np.uint8)\n    return result\n\n\n# ═══════════════════════════════════════════════════════════════════\n# SRM filters  (unchanged)\n# ═══════════════════════════════════════════════════════════════════\n\ndef _srm_kernels():\n    F=[]; z=lambda: np.zeros((5,5))\n    f=z(); f[2,1:4]=[-1,2,-1];      F.append(f/2)\n    f=z(); f[1:4,2]=[-1,2,-1];      F.append(f/2)\n    f=z(); f[2,0:5]=[-1,2,-6,2,-1]; F.append(f/4)\n    f=z(); f[:,2]  =[-1,2,-6,2,-1]; F.append(f/4)\n    f=z(); f[1,1]=f[3,3]=-1; f[2,2]=2; F.append(f/2)\n    f=z(); f[1,3]=f[3,1]=-1; f[2,2]=2; F.append(f/2)\n    f=z(); f[2,1:5]=[-1,3,-3,1];    F.append(f/2)\n    f=z(); f[1:5,2]=[-1,3,-3,1];    F.append(f/2)\n    F.append(np.array([[0,0,-1,0,0],[0,0,2,0,0],[-1,2,-4,2,-1],[0,0,2,0,0],[0,0,-1,0,0]])/4)\n    F.append(np.array([[0,0,0,0,0],[0,-1,2,-1,0],[0,2,-4,2,0],[0,-1,2,-1,0],[0,0,0,0,0]])/4)\n    kb=np.array([5,5,5,-3,0,-3,-3,-3,-3],dtype=float).reshape(3,3)/15\n    for r in range(8):\n        f=z(); f[1:4,1:4]=np.rot90(kb,r%4)*(1 if r<4 else -1); F.append(f)\n    for dy,dx in [(0,1),(0,-1),(1,0),(-1,0),(1,1),(-1,-1),(1,-1),(-1,1)]:\n        f=z(); f[2,2]=1; f[2+dy,2+dx]=-1; F.append(f)\n    while len(F)<30:\n        f=z(); f[2,2]=1; f[2,1]=f[2,3]=f[1,2]=f[3,2]=-0.25; F.append(f)\n    return torch.tensor(np.stack(F[:30])[:,None], dtype=torch.float32)\n\nclass SRM(nn.Module):\n    def __init__(self):\n        super().__init__()\n        self.register_buffer('w', _srm_kernels())\n    def forward(self, x):\n        x = x.mean(dim=1, keepdim=True)\n        return torch.tanh(F.conv2d(x*255, self.w, padding=2)/3)\n\n\n# ═══════════════════════════════════════════════════════════════════\n# SRNet  (architecture unchanged)\n# ═══════════════════════════════════════════════════════════════════\n\nclass SE(nn.Module):\n    def __init__(self, c, r=8):\n        super().__init__()\n        self.fc = nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(),\n                                nn.Linear(c,c//r,False), nn.ReLU(True),\n                                nn.Linear(c//r,c,False), nn.Sigmoid())\n    def forward(self,x): return x * self.fc(x).view(x.size(0),-1,1,1)\n\nclass Blk(nn.Module):\n    def __init__(self, ci, co, se=False):\n        super().__init__()\n        self.net  = nn.Sequential(\n            nn.Conv2d(ci,co,3,padding=1,bias=False), nn.BatchNorm2d(co), nn.ReLU(True),\n            nn.Conv2d(co,co,3,padding=1,bias=False), nn.BatchNorm2d(co))\n        self.skip = nn.Identity() if ci==co else nn.Conv2d(ci,co,1,bias=False)\n        self.se   = SE(co) if se else nn.Identity()\n    def forward(self,x): return F.relu(self.se(self.net(x))+self.skip(x),True)\n\ndef _dn(ci,co):\n    return nn.Sequential(nn.Conv2d(ci,co,3,stride=2,padding=1,bias=False),\n                         nn.BatchNorm2d(co), nn.ReLU(True))\n\nclass SRNet(nn.Module):\n    def __init__(self, nc=2, drop=0.4):\n        super().__init__()\n        self.srm   = SRM()\n        self.entry = nn.Sequential(nn.Conv2d(30,64,3,padding=1,bias=False),\n                                   nn.BatchNorm2d(64), nn.ReLU(True))\n        self.s1=nn.Sequential(Blk(64,64),Blk(64,64));               self.d1=_dn(64,64)\n        self.s2=nn.Sequential(Blk(64,128),Blk(128,128));            self.d2=_dn(128,128)\n        self.s3=nn.Sequential(Blk(128,256,True),Blk(256,256,True)); self.d3=_dn(256,256)\n        self.s4=nn.Sequential(Blk(256,512,True),Blk(512,512,True))\n        self.head=nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(),\n                                nn.Dropout(drop), nn.Linear(512,nc))\n    def forward(self,x):\n        x=self.entry(self.srm(x))\n        x=self.d1(self.s1(x)); x=self.d2(self.s2(x))\n        x=self.d3(self.s3(x)); return self.head(self.s4(x))\n\n\n# ═══════════════════════════════════════════════════════════════════\n# DATASET  —  on-the-fly embedding, variable rate\n#             [FIX C] stronger + more realistic augmentation\n# ═══════════════════════════════════════════════════════════════════\n\ndef _paths(folder, limit=None):\n    if not folder or not os.path.exists(folder): return []\n    exts  = {'.jpg','.jpeg','.png','.bmp','.pgm'}\n    paths = sorted([str(p) for p in Path(folder).rglob('*')\n                    if p.suffix.lower() in exts])\n    return paths[:limit] if limit else paths\n\ndef split_paths(cover_dir, dct_dir, n_train, n_val, n_test):\n    \"\"\"Non-overlapping train / val / test split matched by filename stem.\"\"\"\n    covers    = _paths(cover_dir)\n    cover_map = {Path(p).stem: p for p in covers}\n\n    if dct_dir and os.path.exists(dct_dir):\n        dcts    = _paths(dct_dir)\n        dct_map = {Path(p).stem: p for p in dcts}\n        common  = sorted(set(cover_map.keys()) & set(dct_map.keys()))\n    else:\n        common  = sorted(cover_map.keys())\n        dct_map = {}\n\n    random.shuffle(common)\n    total_need = n_train + n_val + n_test\n    if len(common) < total_need:\n        # Proportional fallback\n        n_train = int(len(common) * 0.75)\n        n_val   = int(len(common) * 0.15)\n        n_test  = len(common) - n_train - n_val\n        print(f\"  ⚠ Only {len(common)} pairs → {n_train} train / {n_val} val / {n_test} test\")\n\n    def _pair(stems):\n        cvs = [cover_map[s] for s in stems]\n        dts = [dct_map.get(s) for s in stems]\n        return cvs, dts\n\n    tr_cov, tr_dct = _pair(common[:n_train])\n    va_cov, va_dct = _pair(common[n_train:n_train+n_val])\n    te_cov, te_dct = _pair(common[n_train+n_val:n_train+n_val+n_test])\n    return tr_cov, tr_dct, va_cov, va_dct, te_cov, te_dct\n\n\ndef _jpeg_compress(arr_uint8: np.ndarray, quality: int) -> np.ndarray:\n    \"\"\"Round-trip through JPEG compression (PIL → BytesIO → PIL → numpy).\"\"\"\n    pil = Image.fromarray(arr_uint8, mode='YCbCr')\n    buf = io.BytesIO()\n    pil.save(buf, format='JPEG', quality=quality)\n    buf.seek(0)\n    return np.array(Image.open(buf).convert('YCbCr'))\n\n\ndef _chip(arr, size, train):\n    \"\"\"\n    Crop + flip augmentation for training.\n    [FIX C] Stronger Gaussian noise (σ=3, p=0.50) + JPEG recompression (p=0.40).\n    \"\"\"\n    h, w = arr.shape[0], arr.shape[1]\n    if train and h >= size and w >= size:\n        y = random.randint(0, h-size)\n        x = random.randint(0, w-size)\n        arr = arr[y:y+size, x:x+size] if arr.ndim==2 else arr[y:y+size, x:x+size,:]\n        if random.random() < .5:\n            arr = (arr[:,::-1] if arr.ndim==2 else arr[:,::-1,:]).copy()\n        if random.random() < .5:\n            arr = (arr[::-1]   if arr.ndim==2 else arr[::-1,:,:]).copy()\n\n        # [FIX C] Gaussian noise — σ raised to 3.0, probability to 0.50\n        if random.random() < 0.50:\n            noise = np.random.normal(0, 3.0, arr.shape).astype(np.float32)\n            arr   = np.clip(arr.astype(np.float32) + noise, 0, 255).astype(np.uint8)\n\n        # [FIX C] JPEG recompression — realistic post-processing distortion\n        if random.random() < 0.40:\n            quality = random.randint(70, 95)\n            arr     = _jpeg_compress(arr, quality)\n\n    else:\n        h0, w0 = (h-size)//2, (w-size)//2\n        if h >= size and w >= size:\n            arr = arr[h0:h0+size, w0:w0+size] if arr.ndim==2 \\\n                  else arr[h0:h0+size, w0:w0+size,:]\n        else:\n            arr = np.array(Image.fromarray(arr).resize((size,size), Image.BILINEAR))\n    return arr\n\ndef _to_tensor(arr):\n    t = torch.from_numpy(arr.astype(np.float32)/255).permute(2,0,1)\n    return (t - 0.5) / 0.5\n\nclass StegoDataset(Dataset):\n    \"\"\"\n    Binary: 0=COVER, 1=DCT_STEGO.  idx%2→label, idx//2→image.\n    On-the-fly embedding with variable rate (train) / fixed 0.3 (val/test).\n    \"\"\"\n    def __init__(self, cover_paths, dct_paths, size=256, train=True,\n                 name=\"DATASET\", use_premade=False):\n        self.covers      = cover_paths\n        self.dcts        = dct_paths\n        self.size        = size\n        self.train       = train\n        self.use_premade = use_premade and any(p is not None for p in (dct_paths or []))\n        src = \"pre-made DCT_DIR\" if self.use_premade else \"on-the-fly embed_dct() [variable rate]\"\n        print(f\"  [{name}] covers={len(cover_paths)} | stego={src} \"\n              f\"| total samples={len(cover_paths)*2:,}\")\n\n    def __len__(self): return len(self.covers) * 2\n\n    def __getitem__(self, idx):\n        lbl = idx % 2\n        ii  = idx // 2\n\n        if lbl == 0:\n            ycbcr = np.array(Image.open(self.covers[ii]).convert('YCbCr'))\n            img   = _chip(ycbcr, self.size, self.train)\n        else:\n            if self.use_premade and self.dcts and self.dcts[ii] is not None:\n                ycbcr = np.array(Image.open(self.dcts[ii]).convert('YCbCr'))\n            else:\n                ycbcr   = np.array(Image.open(self.covers[ii]).convert('YCbCr'))\n                Y       = ycbcr[:,:,0]\n                rate    = random.choice(EMBED_RATES) if self.train else 0.3\n                Y_stego = embed_dct(Y, rate=rate)\n                ycbcr   = np.stack([Y_stego, ycbcr[:,:,1], ycbcr[:,:,2]], axis=2)\n            img = _chip(ycbcr, self.size, self.train)\n\n        return _to_tensor(img), lbl\n\n\n# ═══════════════════════════════════════════════════════════════════\n# TRAIN / EVAL\n# ═══════════════════════════════════════════════════════════════════\n\ndef _epoch(model, loader, crit, opt, scaler, device, train, ep, epochs):\n    model.train(train)\n    ls=cor=tot=0\n    phase    = \"TRAIN\" if train else \"VAL  \"\n    iterator = (tqdm(loader, desc=f\"  Ep{ep:02d}/{epochs} {phase}\",\n                     ncols=90, leave=False) if HAS_TQDM else loader)\n    ctx = torch.enable_grad if train else torch.no_grad\n    with ctx():\n        for bi,(x,y) in enumerate(iterator):\n            x,y = x.to(device), y.to(device)\n            with torch.amp.autocast('cuda', enabled=device.type=='cuda'):\n                out  = model(x)\n                loss = crit(out, y)\n            if train:\n                opt.zero_grad()\n                scaler.scale(loss).backward()\n                scaler.unscale_(opt)\n                nn.utils.clip_grad_norm_(model.parameters(), 3)\n                scaler.step(opt); scaler.update()\n            ls  += loss.item(); pred = out.argmax(1)\n            tot += y.size(0);   cor += (pred==y).sum().item()\n            if HAS_TQDM:\n                iterator.set_postfix(loss=f\"{ls/(bi+1):.3f}\", acc=f\"{100*cor/tot:.1f}%\")\n    return ls/len(loader), 100*cor/tot\n\ndef _eval(model, loader, crit, device):\n    model.eval(); ls=cor=tot=0; ps=[]; ys=[]\n    with torch.no_grad():\n        for x,y in loader:\n            x,y=x.to(device),y.to(device)\n            with torch.amp.autocast('cuda', enabled=device.type=='cuda'):\n                out=model(x); loss=crit(out,y)\n            ls+=loss.item(); pred=out.argmax(1)\n            tot+=y.size(0); cor+=(pred==y).sum().item()\n            ps.extend(pred.cpu().tolist()); ys.extend(y.cpu().tolist())\n    _,_,f,_ = precision_recall_fscore_support(ys,ps,average='binary',zero_division=0)\n    cm        = confusion_matrix(ys,ps,labels=[0,1])\n    cover_acc = 100*sum(p==l for p,l in zip(ps,ys) if l==0)/max(ys.count(0),1)\n    stego_acc = 100*sum(p==l for p,l in zip(ps,ys) if l==1)/max(ys.count(1),1)\n    return ls/len(loader), 100*cor/tot, cover_acc, stego_acc, f*100, cm\n\n\ndef train_model(model, tr_ld, va_ld, device, epochs=50, lr=1e-4, wd=5e-4,\n                warmup=5, patience=7, save='best.pth'):\n\n    crit = nn.CrossEntropyLoss()\n    opt  = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=wd)\n\n    # Cosine schedule with linear warmup\n    def lr_lambda(ep):\n        if ep < warmup: return (ep+1)/warmup\n        progress = (ep-warmup)/(epochs-warmup)\n        return 0.01 + 0.99*0.5*(1+np.cos(np.pi*progress))\n    sch = torch.optim.lr_scheduler.LambdaLR(opt, lr_lambda)\n    sc  = torch.amp.GradScaler('cuda', enabled=device.type=='cuda')\n\n    best_acc       = best_stego = 0.\n    collapse_count = 0\n    best_val_loss  = float('inf')\n    no_improve     = 0\n\n    print(f\"\\n{'='*62}\")\n    print(f\"  Training {epochs} epochs | {len(tr_ld)} batches/ep | {device}\")\n    print(f\"  LR={lr:.0e}  WD={wd:.0e}  Warmup={warmup}ep  Patience={patience}\")\n    print(f\"  Dropout=0.4  Rates={EMBED_RATES}  CrossEntropyLoss\")\n    print(f\"{'='*62}\\n\")\n\n    for ep in range(1, epochs+1):\n        t0 = time.time()\n        print(f\"┌─ Epoch {ep}/{epochs} {'─'*46}\")\n\n        tr_loss, tr_acc = _epoch(model,tr_ld,crit,opt,sc,device,True,ep,epochs)\n        va_loss, va_acc, cov_acc, stg_acc, f1, cm = _eval(model,va_ld,crit,device)\n        sch.step()\n        elapsed = time.time()-t0\n\n        print(f\"│  Loss   : train={tr_loss:.4f}  val={va_loss:.4f}\")\n        print(f\"│  Acc    : train={tr_acc:.1f}%  val={va_acc:.1f}%  F1={f1:.1f}%\")\n        print(f\"│  Class  : COVER={cov_acc:.1f}%  DCT_STEGO={stg_acc:.1f}%\")\n        print(f\"│  LR     : {sch.get_last_lr()[0]:.2e}  |  time: {elapsed:.0f}s\")\n\n        # [FIX D] Collapse guard — restore *scheduler's* current LR, not initial LR\n        if stg_acc < 15.0:\n            collapse_count += 1\n            print(f\"│  ⚠ Collapse warning ({collapse_count}/3)  stego_acc={stg_acc:.1f}%\")\n            if collapse_count >= 3:\n                current_lr = sch.get_last_lr()[0]   # <- FIX D: use scheduler's LR\n                print(f\"│  🔄 LR nudge: temporarily ×2 from {current_lr:.2e}\")\n                for g in opt.param_groups: g['lr'] = current_lr * 2\n                collapse_count = 0\n        else:\n            collapse_count = 0\n\n        # Checkpointing\n        saved=[]\n        if va_acc > best_acc and stg_acc > 20.0:\n            best_acc = va_acc\n            torch.save({'ep':ep,'state':model.state_dict(),\n                        'acc':va_acc,'stego_acc':stg_acc,'cm':cm.tolist()}, save)\n            saved.append(f\"best overall ({va_acc:.2f}%)\")\n        if stg_acc > best_stego:\n            best_stego = stg_acc\n            torch.save({'ep':ep,'state':model.state_dict(),\n                        'acc':va_acc,'stego_acc':stg_acc,'cm':cm.tolist()},\n                       save.replace('.pth','_bestStego.pth'))\n            saved.append(f\"best stego ({stg_acc:.2f}%)\")\n        if saved: print(f\"│  ✓ Saved: {' | '.join(saved)}\")\n\n        # Early stopping on val loss\n        if va_loss < best_val_loss - 1e-4:\n            best_val_loss = va_loss\n            no_improve    = 0\n        else:\n            no_improve += 1\n            print(f\"│  Early stop counter: {no_improve}/{patience}\")\n            if no_improve >= patience:\n                print(f\"│  ⛔ Early stopping triggered at epoch {ep}\")\n                print(f\"└─{'─'*55}\\n\")\n                break\n\n        print(f\"└─{'─'*55}\\n\")\n\n    # Confusion matrix from best checkpoint\n    best_path = save.replace('.pth','_bestStego.pth')\n    ckpt_path = best_path if os.path.exists(best_path) else save\n    if os.path.exists(ckpt_path):\n        cm = np.array(torch.load(ckpt_path, map_location='cpu')['cm'])\n        print(f\"\\n{'='*62}\")\n        print(f\"  CONFUSION MATRIX  (val set — rows=true, cols=predicted)\")\n        print(f\"{'':>14}{'COVER':>14}{'DCT_STEGO':>14}\")\n        for i,n in enumerate(CLASS_NAMES):\n            print(f\"{n:>14}\" + \"\".join(f\"{cm[i,j]:>14}\" for j in range(2)))\n    print(f\"\\n  Best val acc={best_acc:.2f}%  |  Best stego acc={best_stego:.2f}%\")\n    print(f\"{'='*62}\\n\")\n    return best_acc, ckpt_path\n\n\n# ═══════════════════════════════════════════════════════════════════\n# [FIX E] HELD-OUT TEST EVALUATION\n# ═══════════════════════════════════════════════════════════════════\n\ndef evaluate_test(model, te_ld, device, label=\"TEST\"):\n    \"\"\"Run _eval once on the held-out test split and pretty-print results.\"\"\"\n    crit = nn.CrossEntropyLoss()\n    te_loss, te_acc, cov_acc, stg_acc, f1, cm = _eval(model, te_ld, crit, device)\n    print(f\"\\n{'='*62}\")\n    print(f\"  {label} SET RESULTS  (never seen during training or val selection)\")\n    print(f\"  Loss={te_loss:.4f}  Acc={te_acc:.2f}%  F1={f1:.2f}%\")\n    print(f\"  COVER acc={cov_acc:.2f}%  |  DCT_STEGO acc={stg_acc:.2f}%\")\n    print(f\"\\n  CONFUSION MATRIX  (rows=true, cols=predicted)\")\n    print(f\"{'':>14}{'COVER':>14}{'DCT_STEGO':>14}\")\n    for i,n in enumerate(CLASS_NAMES):\n        print(f\"{n:>14}\" + \"\".join(f\"{cm[i,j]:>14}\" for j in range(2)))\n    print(f\"{'='*62}\\n\")\n    return te_acc, stg_acc, f1\n\n\n# ═══════════════════════════════════════════════════════════════════\n# DETECT  (unchanged)\n# ═══════════════════════════════════════════════════════════════════\n\n@torch.no_grad()\ndef detect(model, image, device, size=256):\n    model.eval()\n    if isinstance(image, str):          pil = Image.open(image)\n    elif isinstance(image, np.ndarray): pil = Image.fromarray(image)\n    else:                               pil = image\n    arr   = np.array(pil.convert('YCbCr').resize((size,size), Image.BILINEAR))\n    t     = ((torch.from_numpy(arr.astype(np.float32)/255).permute(2,0,1)-0.5)/0.5)\n    probs = torch.softmax(model(t.unsqueeze(0).to(device)),1)[0].cpu().tolist()\n    pred  = int(np.argmax(probs))\n    print(f\"\\n{'─'*40}\")\n    print(f\"  Prediction : {CLASS_NAMES[pred]}  ({max(probs)*100:.1f}% confidence)\")\n    print(f\"{'─'*40}\")\n    for i,n in enumerate(CLASS_NAMES):\n        print(f\"  {n:<14} {probs[i]*100:5.1f}%  {'█'*int(probs[i]*30)}\")\n    print(f\"{'─'*40}\\n\")\n    return CLASS_NAMES[pred], {n:p for n,p in zip(CLASS_NAMES,probs)}\n\n\n# ═══════════════════════════════════════════════════════════════════\n# SEED + MAIN\n# ═══════════════════════════════════════════════════════════════════\n\ndef set_seed(s=42):\n    random.seed(s); np.random.seed(s)\n    torch.manual_seed(s); torch.cuda.manual_seed_all(s)\n\nif __name__ == '__main__':\n\n    # ╔══════════════════════════════════════════════╗\n    # ║                  CONFIG                      ║\n    # ╠══════════════════════════════════════════════╣\n    COVER_DIR   = '/kaggle/input/datasets/mdezazulhaquemahi/analysis/steganalysis/Train'\n    DCT_DIR     = '/kaggle/input/datasets/mdezazulhaquemahi/dct213/DCT2'\n    TEST_IMAGE  = None\n    NUM_TRAIN   = 5000\n    NUM_VAL     = 800\n    NUM_TEST    = 700           # [FIX E] held-out test split\n    IMAGE_SIZE  = 256\n    BATCH_SIZE  = 16\n    EPOCHS      = 50            # [FIX A] raised from 10; early stopping still guards\n    LR          = 1e-4\n    WEIGHT_DEC  = 5e-4\n    WARMUP_EP   = 5\n    PATIENCE    = 7\n    USE_PREMADE = False\n    SAVE        = 'steganalyzer_v8.pth'\n    # ╚══════════════════════════════════════════════╝\n\n    set_seed(42)\n    DEV = torch.device('cuda' if torch.cuda.is_available() else 'cpu')\n\n    print(f\"\\n{'='*62}\")\n    print(f\"  STEGANALYZER v8  —  {' / '.join(CLASS_NAMES)}  (binary)\")\n    print(f\"{'='*62}\")\n    print(f\"  Device : {DEV}\" + (f\" ({torch.cuda.get_device_name(0)})\" if DEV.type=='cuda' else \"\"))\n    print(f\"  Size   : {IMAGE_SIZE}×{IMAGE_SIZE}  |  Batch: {BATCH_SIZE}  |  Epochs: {EPOCHS}\")\n    print(f\"  tqdm   : {'yes' if HAS_TQDM else 'no — pip install tqdm'}\")\n    print(f\"\\n  Splitting dataset (no train/val/test overlap)...\")\n\n    tr_covers, tr_dcts, va_covers, va_dcts, te_covers, te_dcts = split_paths(\n        COVER_DIR, DCT_DIR, NUM_TRAIN, NUM_VAL, NUM_TEST)\n\n    print(f\"  Loading datasets...\")\n    tr_ds = StegoDataset(tr_covers, tr_dcts, IMAGE_SIZE, train=True,\n                         name=\"TRAIN\", use_premade=USE_PREMADE)\n    va_ds = StegoDataset(va_covers, va_dcts, IMAGE_SIZE, train=False,\n                         name=\"VAL  \", use_premade=USE_PREMADE)\n    te_ds = StegoDataset(te_covers, te_dcts, IMAGE_SIZE, train=False,\n                         name=\"TEST \", use_premade=USE_PREMADE)\n    print(f\"  Batches : train={len(tr_ds)//BATCH_SIZE}  \"\n          f\"val={len(va_ds)//BATCH_SIZE}  test={len(te_ds)//BATCH_SIZE}\")\n\n    tr_ld = DataLoader(tr_ds, BATCH_SIZE, shuffle=True,\n                       num_workers=2, pin_memory=True)\n    va_ld = DataLoader(va_ds, BATCH_SIZE, shuffle=False,\n                       num_workers=2, pin_memory=True)\n    te_ld = DataLoader(te_ds, BATCH_SIZE, shuffle=False,\n                       num_workers=2, pin_memory=True)\n\n    model  = SRNet(nc=2, drop=0.4).to(DEV)\n    params = sum(p.numel() for p in model.parameters())\n    print(f\"  Model  : SRNet  ({params:,} params)\\n\")\n\n    _, ckpt_path = train_model(model, tr_ld, va_ld, DEV,\n                               epochs=EPOCHS, lr=LR, wd=WEIGHT_DEC,\n                               warmup=WARMUP_EP, patience=PATIENCE, save=SAVE)\n\n    # Load best checkpoint then run held-out test\n    if os.path.exists(ckpt_path):\n        ckpt  = torch.load(ckpt_path, map_location=DEV)\n        state = ckpt['state'] if 'state' in ckpt else ckpt\n        model.load_state_dict(state)\n        print(f\"  Loaded: {ckpt_path}\")\n\n    # [FIX E] Final unbiased evaluation on the held-out test set\n    evaluate_test(model, te_ld, DEV)\n\n    if TEST_IMAGE and os.path.exists(TEST_IMAGE):\n        detect(model, TEST_IMAGE, DEV, IMAGE_SIZE)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-28T07:55:19.123376Z","iopub.execute_input":"2026-04-28T07:55:19.123746Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}