Victor Phan
275 lines
9.2 KiB
Python
275 lines
9.2 KiB
Python
"""
|
|
PyTorch 1D CNN Model for Land Use Classification
|
|
Dùng cho 3 channels: NDVI, VH, VV (13 time steps)
|
|
"""
|
|
|
|
import torch
|
|
import torch.nn as nn
|
|
import torch.optim as optim
|
|
from torch.utils.data import Dataset, DataLoader
|
|
import numpy as np
|
|
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, confusion_matrix
|
|
import matplotlib.pyplot as plt
|
|
|
|
|
|
class TimeSeriesDataset(Dataset):
|
|
"""Custom Dataset for time series data"""
|
|
def __init__(self, X, y):
|
|
self.X = torch.FloatTensor(X)
|
|
self.y = torch.LongTensor(y)
|
|
|
|
def __len__(self):
|
|
return len(self.X)
|
|
|
|
def __getitem__(self, idx):
|
|
return self.X[idx], self.y[idx]
|
|
|
|
|
|
class CNN1D(nn.Module):
|
|
"""1D CNN for time series classification
|
|
|
|
Input: (batch_size, 3, 13)
|
|
- 3 channels: NDVI, VH, VV
|
|
- 13 time steps
|
|
Output: (batch_size, num_classes)
|
|
"""
|
|
def __init__(self, num_classes=8, dropout_rate=0.3):
|
|
super(CNN1D, self).__init__()
|
|
|
|
# 1D Convolutional layers
|
|
self.conv1 = nn.Conv1d(in_channels=3, out_channels=32, kernel_size=3, padding=1)
|
|
self.bn1 = nn.BatchNorm1d(32)
|
|
self.relu1 = nn.ReLU()
|
|
self.pool1 = nn.MaxPool1d(kernel_size=2, stride=2)
|
|
|
|
self.conv2 = nn.Conv1d(in_channels=32, out_channels=64, kernel_size=3, padding=1)
|
|
self.bn2 = nn.BatchNorm1d(64)
|
|
self.relu2 = nn.ReLU()
|
|
self.pool2 = nn.MaxPool1d(kernel_size=2, stride=2)
|
|
|
|
self.conv3 = nn.Conv1d(in_channels=64, out_channels=128, kernel_size=3, padding=1)
|
|
self.bn3 = nn.BatchNorm1d(128)
|
|
self.relu3 = nn.ReLU()
|
|
self.pool3 = nn.MaxPool1d(kernel_size=2, stride=2)
|
|
|
|
# Global average pooling
|
|
self.global_avg_pool = nn.AdaptiveAvgPool1d(1)
|
|
|
|
# Fully connected layers
|
|
self.fc1 = nn.Linear(128, 64)
|
|
self.dropout = nn.Dropout(dropout_rate)
|
|
self.fc2 = nn.Linear(64, num_classes)
|
|
|
|
def forward(self, x):
|
|
# Conv block 1
|
|
x = self.conv1(x)
|
|
x = self.bn1(x)
|
|
x = self.relu1(x)
|
|
x = self.pool1(x)
|
|
|
|
# Conv block 2
|
|
x = self.conv2(x)
|
|
x = self.bn2(x)
|
|
x = self.relu2(x)
|
|
x = self.pool2(x)
|
|
|
|
# Conv block 3
|
|
x = self.conv3(x)
|
|
x = self.bn3(x)
|
|
x = self.relu3(x)
|
|
x = self.pool3(x)
|
|
|
|
# Global average pooling
|
|
x = self.global_avg_pool(x)
|
|
x = x.view(x.size(0), -1)
|
|
|
|
# FC layers
|
|
x = self.fc1(x)
|
|
x = self.dropout(x)
|
|
x = self.fc2(x)
|
|
|
|
return x
|
|
|
|
|
|
class CNNTrainer:
|
|
"""Trainer for PyTorch CNN Model"""
|
|
|
|
def __init__(self, num_classes=8, learning_rate=0.001, device=None):
|
|
self.device = device or torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
|
self.model = CNN1D(num_classes=num_classes).to(self.device)
|
|
self.criterion = nn.CrossEntropyLoss()
|
|
self.optimizer = optim.Adam(self.model.parameters(), lr=learning_rate)
|
|
self.history = {'train_loss': [], 'val_loss': [], 'train_acc': [], 'val_acc': []}
|
|
|
|
print(f"🚀 Model initialized on device: {self.device}")
|
|
print(f" Total parameters: {sum(p.numel() for p in self.model.parameters()):,}")
|
|
|
|
def train_epoch(self, train_loader):
|
|
"""Train one epoch"""
|
|
self.model.train()
|
|
total_loss = 0
|
|
correct = 0
|
|
total = 0
|
|
|
|
for X_batch, y_batch in train_loader:
|
|
X_batch, y_batch = X_batch.to(self.device), y_batch.to(self.device)
|
|
|
|
# Forward pass
|
|
outputs = self.model(X_batch)
|
|
loss = self.criterion(outputs, y_batch)
|
|
|
|
# Backward pass
|
|
self.optimizer.zero_grad()
|
|
loss.backward()
|
|
self.optimizer.step()
|
|
|
|
# Metrics
|
|
total_loss += loss.item()
|
|
_, predicted = torch.max(outputs.data, 1)
|
|
correct += (predicted == y_batch).sum().item()
|
|
total += y_batch.size(0)
|
|
|
|
avg_loss = total_loss / len(train_loader)
|
|
accuracy = correct / total
|
|
return avg_loss, accuracy
|
|
|
|
def validate(self, val_loader):
|
|
"""Validate model"""
|
|
self.model.eval()
|
|
total_loss = 0
|
|
correct = 0
|
|
total = 0
|
|
|
|
with torch.no_grad():
|
|
for X_batch, y_batch in val_loader:
|
|
X_batch, y_batch = X_batch.to(self.device), y_batch.to(self.device)
|
|
|
|
outputs = self.model(X_batch)
|
|
loss = self.criterion(outputs, y_batch)
|
|
|
|
total_loss += loss.item()
|
|
_, predicted = torch.max(outputs.data, 1)
|
|
correct += (predicted == y_batch).sum().item()
|
|
total += y_batch.size(0)
|
|
|
|
avg_loss = total_loss / len(val_loader)
|
|
accuracy = correct / total
|
|
return avg_loss, accuracy
|
|
|
|
def fit(self, X_train, y_train, X_val, y_val, epochs=50, batch_size=32, verbose=True):
|
|
"""Train model with validation"""
|
|
train_dataset = TimeSeriesDataset(X_train, y_train)
|
|
val_dataset = TimeSeriesDataset(X_val, y_val)
|
|
|
|
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
|
|
val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)
|
|
|
|
print(f"\n📊 Training start:")
|
|
print(f" Train samples: {len(X_train)}")
|
|
print(f" Val samples: {len(X_val)}")
|
|
print(f" Batch size: {batch_size}")
|
|
print(f" Epochs: {epochs}\n")
|
|
|
|
for epoch in range(epochs):
|
|
train_loss, train_acc = self.train_epoch(train_loader)
|
|
val_loss, val_acc = self.validate(val_loader)
|
|
|
|
self.history['train_loss'].append(train_loss)
|
|
self.history['val_loss'].append(val_loss)
|
|
self.history['train_acc'].append(train_acc)
|
|
self.history['val_acc'].append(val_acc)
|
|
|
|
if verbose and (epoch + 1) % 10 == 0:
|
|
print(f"Epoch [{epoch+1}/{epochs}] "
|
|
f"Train Loss: {train_loss:.4f}, Acc: {train_acc:.4f} | "
|
|
f"Val Loss: {val_loss:.4f}, Acc: {val_acc:.4f}")
|
|
|
|
print(f"\n✅ Training completed!")
|
|
print(f" Final Train Acc: {train_acc:.4f}")
|
|
print(f" Final Val Acc: {val_acc:.4f}")
|
|
|
|
def predict(self, X_test):
|
|
"""Predict on test data"""
|
|
self.model.eval()
|
|
X_test = torch.FloatTensor(X_test).to(self.device)
|
|
|
|
with torch.no_grad():
|
|
outputs = self.model(X_test)
|
|
_, predictions = torch.max(outputs, 1)
|
|
|
|
return predictions.cpu().numpy()
|
|
|
|
def evaluate(self, X_test, y_test):
|
|
"""Evaluate on test data"""
|
|
y_pred = self.predict(X_test)
|
|
|
|
accuracy = accuracy_score(y_test, y_pred)
|
|
precision = precision_score(y_test, y_pred, average='weighted', zero_division=0)
|
|
recall = recall_score(y_test, y_pred, average='weighted', zero_division=0)
|
|
f1 = f1_score(y_test, y_pred, average='weighted', zero_division=0)
|
|
|
|
print(f"\n📈 Test Results:")
|
|
print(f" Accuracy: {accuracy:.4f}")
|
|
print(f" Precision: {precision:.4f}")
|
|
print(f" Recall: {recall:.4f}")
|
|
print(f" F1-Score: {f1:.4f}")
|
|
|
|
return {
|
|
'accuracy': accuracy,
|
|
'precision': precision,
|
|
'recall': recall,
|
|
'f1': f1,
|
|
'predictions': y_pred,
|
|
'confusion_matrix': confusion_matrix(y_test, y_pred)
|
|
}
|
|
|
|
def plot_history(self):
|
|
"""Plot training history"""
|
|
fig, axes = plt.subplots(1, 2, figsize=(12, 4))
|
|
|
|
# Loss
|
|
axes[0].plot(self.history['train_loss'], label='Train Loss')
|
|
axes[0].plot(self.history['val_loss'], label='Val Loss')
|
|
axes[0].set_xlabel('Epoch')
|
|
axes[0].set_ylabel('Loss')
|
|
axes[0].set_title('Training and Validation Loss')
|
|
axes[0].legend()
|
|
axes[0].grid(True)
|
|
|
|
# Accuracy
|
|
axes[1].plot(self.history['train_acc'], label='Train Acc')
|
|
axes[1].plot(self.history['val_acc'], label='Val Acc')
|
|
axes[1].set_xlabel('Epoch')
|
|
axes[1].set_ylabel('Accuracy')
|
|
axes[1].set_title('Training and Validation Accuracy')
|
|
axes[1].legend()
|
|
axes[1].grid(True)
|
|
|
|
plt.tight_layout()
|
|
plt.show()
|
|
|
|
def save(self, filepath):
|
|
"""Save model"""
|
|
torch.save(self.model.state_dict(), filepath)
|
|
print(f"✅ Model saved to {filepath}")
|
|
|
|
def load(self, filepath):
|
|
"""Load model"""
|
|
self.model.load_state_dict(torch.load(filepath, map_location=self.device))
|
|
print(f"✅ Model loaded from {filepath}")
|
|
|
|
|
|
def reshape_for_cnn(X):
|
|
"""Reshape data for CNN
|
|
|
|
Input: (n_samples, n_features) where n_features = 13*3 = 39 (13 timesteps x 3 channels)
|
|
Output: (n_samples, 3, 13) - (batch, channels, timesteps)
|
|
"""
|
|
n_samples = X.shape[0]
|
|
n_timesteps = 13
|
|
n_channels = 3
|
|
|
|
# Reshape: (n_samples, 39) -> (n_samples, 3, 13)
|
|
X_cnn = X.reshape(n_samples, n_channels, n_timesteps)
|
|
return X_cnn
|