Project: Image Classification

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as T
from torch.utils.data import DataLoader
import numpy as np

print("=" * 60)
print("🖼️  مشروع: تصنيف صور CIFAR-10")
print("=" * 60)

device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"Device: {device}")

CLASSES = ["airplane","automobile","bird","cat","deer",
           "dog","frog","horse","ship","truck"]

# ─────────────────────────────────────────
# 1. Transforms مع Data Augmentation
# ─────────────────────────────────────────
MEAN = [0.4914, 0.4822, 0.4465]
STD  = [0.2023, 0.1994, 0.2010]

train_transform = T.Compose([
    T.RandomHorizontalFlip(p=0.5),
    T.RandomCrop(32, padding=4),
    T.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.2),
    T.ToTensor(),
    T.Normalize(MEAN, STD),
])
val_transform = T.Compose([
    T.ToTensor(),
    T.Normalize(MEAN, STD),
])

print("\n1. تحميل CIFAR-10...")
train_set = torchvision.datasets.CIFAR10("./data", train=True,  transform=train_transform, download=True)
val_set   = torchvision.datasets.CIFAR10("./data", train=False, transform=val_transform,   download=True)

train_loader = DataLoader(train_set, batch_size=128, shuffle=True,  num_workers=0, pin_memory=True)
val_loader   = DataLoader(val_set,   batch_size=256, shuffle=False, num_workers=0, pin_memory=True)
print(f"   Train: {len(train_set):,} | Val: {len(val_set):,}")

# ─────────────────────────────────────────
# 2. ResNet-Inspired Architecture
# ─────────────────────────────────────────
class ResBlock(nn.Module):
    """Residual Block: التمرير المباشر يحلّ Vanishing Gradient"""
    def __init__(self, ch: int):
        super().__init__()
        self.conv = nn.Sequential(
            nn.Conv2d(ch, ch, 3, padding=1, bias=False),
            nn.BatchNorm2d(ch), nn.ReLU(inplace=True),
            nn.Conv2d(ch, ch, 3, padding=1, bias=False),
            nn.BatchNorm2d(ch),
        )
        self.relu = nn.ReLU(inplace=True)

    def forward(self, x):
        return self.relu(x + self.conv(x))  # ← Residual connection

class ResNetSmall(nn.Module):
    def __init__(self, num_classes: int = 10):
        super().__init__()
        self.stem = nn.Sequential(
            nn.Conv2d(3, 64, 3, padding=1, bias=False),
            nn.BatchNorm2d(64), nn.ReLU(inplace=True),
        )
        self.layer1 = nn.Sequential(ResBlock(64), ResBlock(64))
        self.down1  = nn.Sequential(
            nn.Conv2d(64, 128, 3, stride=2, padding=1, bias=False),
            nn.BatchNorm2d(128), nn.ReLU(inplace=True),
        )
        self.layer2 = nn.Sequential(ResBlock(128), ResBlock(128))
        self.pool   = nn.AdaptiveAvgPool2d(1)
        self.fc     = nn.Linear(128, num_classes)

    def forward(self, x):
        x = self.stem(x)
        x = self.layer1(x)
        x = self.down1(x)
        x = self.layer2(x)
        x = self.pool(x).flatten(1)
        return self.fc(x)

model  = ResNetSmall().to(device)
params = sum(p.numel() for p in model.parameters())
print(f"\n2. النموذج: {params:,} معامل")

# ─────────────────────────────────────────
# 3. Training Setup
# ─────────────────────────────────────────
criterion = nn.CrossEntropyLoss(label_smoothing=0.1)
optimizer = optim.SGD(model.parameters(), lr=0.1,
                       momentum=0.9, weight_decay=5e-4, nesterov=True)
scheduler = optim.lr_scheduler.OneCycleLR(
    optimizer, max_lr=0.1,
    epochs=20, steps_per_epoch=len(train_loader)
)

# ─────────────────────────────────────────
# 4. Training Loop
# ─────────────────────────────────────────
def run_epoch(model, loader, opt=None, crit=None):
    training = opt is not None
    model.train() if training else model.eval()
    loss_sum = correct = total = 0
    ctx = torch.enable_grad() if training else torch.no_grad()
    with ctx:
        for X, y in loader:
            X, y = X.to(device), y.to(device)
            out  = model(X)
            loss = crit(out, y)
            if training:
                opt.zero_grad()
                loss.backward()
                opt.step()
                scheduler.step()
            loss_sum += loss.item() * len(y)
            correct  += (out.argmax(1) == y).sum().item()
            total    += len(y)
    return loss_sum / total, correct / total

print("\n3. التدريب (20 epoch):")
print(f"  {'EP':4s} | {'TrLoss':8s} | {'TrAcc':7s} | {'VlAcc':7s} | LR")
print("  " + "-" * 50)

best_acc = 0
for ep in range(1, 21):
    tr_loss, tr_acc = run_epoch(model, train_loader, optimizer, criterion)
    _,       vl_acc = run_epoch(model, val_loader)
    lr = optimizer.param_groups[0]["lr"]

    if vl_acc > best_acc:
        best_acc = vl_acc
        torch.save(model.state_dict(), "best_cifar.pt")

    if ep % 5 == 0 or ep == 1:
        print(f"  {ep:4d} | {tr_loss:8.4f} | {tr_acc:7.1%} | {vl_acc:7.1%} | {lr:.5f}")

print(f"\n  أفضل Val Accuracy: {best_acc:.1%}")

# ─────────────────────────────────────────
# 5. Per-Class Accuracy
# ─────────────────────────────────────────
model.load_state_dict(torch.load("best_cifar.pt", map_location=device))
model.eval()
class_correct = [0] * 10
class_total   = [0] * 10
with torch.no_grad():
    for X, y in val_loader:
        X, y = X.to(device), y.to(device)
        preds = model(X).argmax(1)
        for c in range(10):
            mask = (y == c)
            class_correct[c] += (preds[mask] == c).sum().item()
            class_total[c]   += mask.sum().item()

print("\n4. دقة كل فئة:")
for i, cls in enumerate(CLASSES):
    acc = class_correct[i] / class_total[i] if class_total[i] > 0 else 0
    bar = "█" * int(acc * 25)
    print(f"  {cls:12s}: {acc:.1%} {bar}")

print("\n🎉 مبروك! أكملت دورة التعلم العميق")