diff --git a/__pycache__/new_import_ODC.cpython-310.pyc b/__pycache__/new_import_ODC.cpython-310.pyc new file mode 100644 index 0000000..1d30f29 Binary files /dev/null and b/__pycache__/new_import_ODC.cpython-310.pyc differ diff --git a/cnn_model.py b/cnn_model.py new file mode 100644 index 0000000..a648b14 --- /dev/null +++ b/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