Merge pull request 'CNN' (#1) from CNN into main

Reviewed-on: #1

cnn_model.py

@@ -0,0 +1,274 @@
+"""
+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