Transfer Learning: Pre-trained Models & Fine-Tuning

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import models, transforms, datasets
from torch.utils.data import DataLoader
import time
import copy

# === 1. PERSIAPAN DATA ===
# Transformasi untuk training (dengan augmentasi)
train_transform = transforms.Compose([
    transforms.RandomResizedCrop(224),
    transforms.RandomHorizontalFlip(),
    transforms.RandomRotation(15),
    transforms.ColorJitter(brightness=0.2, contrast=0.2),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406],  # ImageNet mean
                         [0.229, 0.224, 0.225])   # ImageNet std
])

# Transformasi untuk validasi (tanpa augmentasi)
val_transform = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406],
                         [0.229, 0.224, 0.225])
])

# Load dataset (ganti dengan path dataset Anda)
# train_dataset = datasets.ImageFolder('data/train', train_transform)
# val_dataset = datasets.ImageFolder('data/val', val_transform)
# train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
# val_loader = DataLoader(val_dataset, batch_size=32)

# === 2. LOAD PRE-TRAINED MODEL ===
model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)

print("=== Arsitektur ResNet50 ===")
print(model)

# === 3. FEATURE EXTRACTION (Strategy 1: Freeze semua, train head) ===

# Bekukan semua parameter
for param in model.parameters():
    param.requires_grad = False

# Ganti classifier head untuk 10 kelas
num_classes = 10
model.fc = nn.Sequential(
    nn.Linear(model.fc.in_features, 512),
    nn.ReLU(),
    nn.Dropout(0.3),
    nn.Linear(512, num_classes)
)

# Hitung parameter yang trainable
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in model.parameters())
print(f"\n=== Feature Extraction Mode ===")
print(f"Total parameters: {total_params:,}")
print(f"Trainable parameters: {trainable_params:,}")
print(f"Frozen parameters: {total_params - trainable_params:,}")
print(f"Trainable %: {trainable_params/total_params*100:.1f}%")

# === 4. FINE-TUNING (Strategy 2: Unfreeze beberapa layer) ===

def unfreeze_layers(model, num_layers_to_unfreeze):
    """Unfreeze layer terakhir dari ResNet"""
    # Freeze semua dulu
    for param in model.parameters():
        param.requires_grad = False
    
    # Unfreeze layer4 dan fc
    children = list(model.children())
    layers_to_unfreeze = children[-num_layers_to_unfreeze:]
    
    for layer in layers_to_unfreeze:
        for param in layer.parameters():
            param.requires_grad = True
    
    trainable = sum(p.numel() for p in model.parameters() if p.requires_grad)
    total = sum(p.numel() for p in model.parameters())
    print(f"\nUnfroze {num_layers_to_unfreeze} layers")
    print(f"Trainable: {trainable:,} / {total:,} ({trainable/total*100:.1f}%)")

# Unfreeze 2 layer terakhir (layer4 + fc)
unfreeze_layers(model, 2)

# === 5. DIFFERENTIAL LEARNING RATES ===
# Layer awal → lr kecil, layer akhir → lr lebih besar
optimizer = optim.Adam([
    {'params': model.layer4.parameters(), 'lr': 1e-4},
    {'params': model.fc.parameters(), 'lr': 1e-3},
], weight_decay=1e-4)

# Scheduler: kurangi lr saat training plateau
scheduler = optim.lr_scheduler.ReduceLROnPlateau(
    optimizer, mode='min', factor=0.5, patience=3, verbose=True
)

criterion = nn.CrossEntropyLoss()

# === 6. TRAINING LOOP ===
def train_model(model, criterion, optimizer, scheduler, num_epochs=25):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = model.to(device)
    
    best_model_wts = copy.deepcopy(model.state_dict())
    best_acc = 0.0
    
    for epoch in range(num_epochs):
        print(f'\nEpoch {epoch+1}/{num_epochs}')
        print('-' * 40)
        
        for phase in ['train', 'val']:
            if phase == 'train':
                model.train()
                dataloader = train_loader
            else:
                model.eval()
                dataloader = val_loader
            
            running_loss = 0.0
            running_corrects = 0
            
            for inputs, labels in dataloader:
                inputs = inputs.to(device)
                labels = labels.to(device)
                
                optimizer.zero_grad()
                
                with torch.set_grad_enabled(phase == 'train'):
                    outputs = model(inputs)
                    _, preds = torch.max(outputs, 1)
                    loss = criterion(outputs, labels)
                    
                    if phase == 'train':
                        loss.backward()
                        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
                        optimizer.step()
                
                running_loss += loss.item() * inputs.size(0)
                running_corrects += torch.sum(preds == labels.data)
            
            epoch_loss = running_loss / len(dataloader.dataset)
            epoch_acc = running_corrects.double() / len(dataloader.dataset)
            
            print(f'{phase} Loss: {epoch_loss:.4f} Acc: {epoch_acc:.4f}')
            
            if phase == 'val':
                scheduler.step(epoch_loss)
                if epoch_acc > best_acc:
                    best_acc = epoch_acc
                    best_model_wts = copy.deepcopy(model.state_dict())
    
    print(f'\nBest val Acc: {best_acc:.4f}')
    model.load_state_dict(best_model_wts)
    return model

# Jalankan training (uncomment jika data tersedia)
# model = train_model(model, criterion, optimizer, scheduler, num_epochs=20)

# === 7. PREDIKSI ===
def predict_image(model, image_path, transform):
    from PIL import Image
    
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.eval()
    
    image = Image.open(image_path).convert('RGB')
    image = transform(image).unsqueeze(0).to(device)
    
    with torch.no_grad():
        output = model(image)
        probabilities = torch.nn.functional.softmax(output[0], dim=0)
        predicted_class = torch.argmax(probabilities).item()
        confidence = probabilities[predicted_class].item()
    
    return predicted_class, confidence, probabilities

print("\n✓ Transfer Learning Pipeline Siap!")
print("  Uncomment fungsi train_model() dan siapkan dataset untuk mulai training.")

import tensorflow as tf
from tensorflow.keras import layers, models
from tensorflow.keras.applications import EfficientNetB0
from tensorflow.keras.preprocessing import image_dataset_from_directory

# === 1. LOAD DATA ===
IMG_SIZE = 224
BATCH_SIZE = 32

# Load dataset (ganti dengan path dataset Anda)
# train_ds = image_dataset_from_directory(
#     'data/train', image_size=(IMG_SIZE, IMG_SIZE),
#     batch_size=BATCH_SIZE, validation_split=0.2,
#     subset='training', seed=42
# )
# val_ds = image_dataset_from_directory(
#     'data/train', image_size=(IMG_SIZE, IMG_SIZE),
#     batch_size=BATCH_SIZE, validation_split=0.2,
#     subset='validation', seed=42
# )

NUM_CLASSES = 10

# === 2. DATA AUGMENTATION ===
data_augmentation = tf.keras.Sequential([
    layers.RandomFlip("horizontal"),
    layers.RandomRotation(0.1),
    layers.RandomZoom(0.1),
    layers.RandomContrast(0.1),
])

# === 3. LOAD PRE-TRAINED EfficientNetB0 ===
base_model = EfficientNetB0(
    include_top=False,           # Tanpa classifier head
    weights='imagenet',          # Pre-trained di ImageNet
    input_shape=(IMG_SIZE, IMG_SIZE, 3)
)

# === 4. FEATURE EXTRACTION (freeze base model) ===
base_model.trainable = False

# Buat model baru
inputs = layers.Input(shape=(IMG_SIZE, IMG_SIZE, 3))
x = data_augmentation(inputs)               # Augmentasi
x = base_model(x, training=False)           # Feature extraction
x = layers.GlobalAveragePooling2D()(x)      # Pooling
x = layers.Dropout(0.3)(x)                  # Regularisasi
x = layers.Dense(256, activation='relu')(x) # Hidden layer
x = layers.Dropout(0.2)(x)
outputs = layers.Dense(NUM_CLASSES, activation='softmax')(x)

model = models.Model(inputs, outputs)

print("=== Model Summary (Feature Extraction) ===")
model.summary()

# === 5. TRAIN FEATURE EXTRACTION ===
model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=1e-3),
    loss='sparse_categorical_crossentropy',
    metrics=['accuracy']
)

# history1 = model.fit(train_ds, validation_data=val_ds, epochs=10)

# === 6. FINE-TUNING (unfreeze beberapa layer) ===
base_model.trainable = True

# Freeze semua kecuali 30 layer terakhir
for layer in base_model.layers[:-30]:
    layer.trainable = False

print(f"\nTrainable layers: {sum(1 for l in model.layers if l.trainable)}")

# Compile ulang dengan learning rate LEBIH KECIL
model.compile(
    optimizer=tf.keras.optimizers.Adam(learning_rate=1e-5),  # LR kecil!
    loss='sparse_categorical_crossentropy',
    metrics=['accuracy']
)

# history2 = model.fit(train_ds, validation_data=val_ds, epochs=20)

# === 7. LEARNING RATE SCHEDULER ===
lr_scheduler = tf.keras.callbacks.ReduceLROnPlateau(
    monitor='val_loss', factor=0.5, patience=3, min_lr=1e-7
)

early_stopping = tf.keras.callbacks.EarlyStopping(
    monitor='val_loss', patience=7, restore_best_weights=True
)

# history2 = model.fit(
#     train_ds, validation_data=val_ds, epochs=50,
#     callbacks=[lr_scheduler, early_stopping]
# )

print("\n✓ TensorFlow Transfer Learning Pipeline Siap!")