feat: add deep learning classifier comparison script for AURIS with multiple architectures

app/training/train_deep_classifiers.py

@@ -0,0 +1,316 @@
+"""
+Deep learning classifier comparison for AURIS.
+
+Trains and evaluates multiple neural network architectures on
+the 47 extracted audio features using stratified k-fold CV.
+
+Architectures:
+  1. Deep MLP (512-256-128-64) with BatchNorm + Dropout
+  2. 1D-CNN on feature vector (treats features as 1D signal)
+  3. Residual MLP (skip connections)
+  4. Attention MLP (self-attention over feature groups)
+
+Usage:
+    python -m app.training.train_deep_classifiers ../DataSet/features.csv
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import sys
+import time
+from pathlib import Path
+
+import numpy as np
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader, TensorDataset
+from sklearn.model_selection import StratifiedKFold
+from sklearn.preprocessing import StandardScaler
+from sklearn.metrics import accuracy_score, roc_auc_score, f1_score, precision_score, recall_score
+
+DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+SEED = 42
+N_FOLDS = 5
+EPOCHS = 100
+PATIENCE = 10
+BATCH_SIZE = 64
+LR = 1e-3
+
+
+def set_seed(seed: int = SEED) -> None:
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed_all(seed)
+
+
+class DeepMLP(nn.Module):
+    def __init__(self, n_features: int) -> None:
+        super().__init__()
+        self.net = nn.Sequential(
+            nn.Linear(n_features, 512),
+            nn.BatchNorm1d(512),
+            nn.ReLU(),
+            nn.Dropout(0.4),
+            nn.Linear(512, 256),
+            nn.BatchNorm1d(256),
+            nn.ReLU(),
+            nn.Dropout(0.3),
+            nn.Linear(256, 128),
+            nn.BatchNorm1d(128),
+            nn.ReLU(),
+            nn.Dropout(0.2),
+            nn.Linear(128, 64),
+            nn.ReLU(),
+            nn.Linear(64, 1),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.net(x).squeeze(-1)
+
+
+class Conv1DClassifier(nn.Module):
+    def __init__(self, n_features: int) -> None:
+        super().__init__()
+        self.conv = nn.Sequential(
+            nn.Conv1d(1, 64, kernel_size=5, padding=2),
+            nn.BatchNorm1d(64),
+            nn.ReLU(),
+            nn.Conv1d(64, 128, kernel_size=3, padding=1),
+            nn.BatchNorm1d(128),
+            nn.ReLU(),
+            nn.AdaptiveAvgPool1d(1),
+        )
+        self.fc = nn.Sequential(
+            nn.Linear(128, 64),
+            nn.ReLU(),
+            nn.Dropout(0.3),
+            nn.Linear(64, 1),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        x = x.unsqueeze(1)  # (B, 1, F)
+        x = self.conv(x).squeeze(-1)  # (B, 128)
+        return self.fc(x).squeeze(-1)
+
+
+class ResidualBlock(nn.Module):
+    def __init__(self, dim: int, dropout: float = 0.2) -> None:
+        super().__init__()
+        self.block = nn.Sequential(
+            nn.Linear(dim, dim),
+            nn.BatchNorm1d(dim),
+            nn.ReLU(),
+            nn.Dropout(dropout),
+            nn.Linear(dim, dim),
+            nn.BatchNorm1d(dim),
+        )
+        self.relu = nn.ReLU()
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.relu(x + self.block(x))
+
+
+class ResidualMLP(nn.Module):
+    def __init__(self, n_features: int) -> None:
+        super().__init__()
+        self.input_proj = nn.Sequential(
+            nn.Linear(n_features, 256),
+            nn.BatchNorm1d(256),
+            nn.ReLU(),
+        )
+        self.res_blocks = nn.Sequential(
+            ResidualBlock(256, 0.3),
+            ResidualBlock(256, 0.2),
+            ResidualBlock(256, 0.1),
+        )
+        self.head = nn.Sequential(
+            nn.Linear(256, 64),
+            nn.ReLU(),
+            nn.Linear(64, 1),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        x = self.input_proj(x)
+        x = self.res_blocks(x)
+        return self.head(x).squeeze(-1)
+
+
+class AttentionMLP(nn.Module):
+    def __init__(self, n_features: int) -> None:
+        super().__init__()
+        self.proj = nn.Linear(n_features, 256)
+        self.attn = nn.MultiheadAttention(256, num_heads=4, batch_first=True)
+        self.norm = nn.LayerNorm(256)
+        self.head = nn.Sequential(
+            nn.Linear(256, 128),
+            nn.ReLU(),
+            nn.Dropout(0.3),
+            nn.Linear(128, 1),
+        )
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        x = self.proj(x)
+        x = x.unsqueeze(1)  # (B, 1, 256)
+        x_chunk = x.expand(-1, 4, -1)  # (B, 4, 256) - create sequence
+        attn_out, _ = self.attn(x_chunk, x_chunk, x_chunk)
+        x = self.norm(attn_out.mean(dim=1))  # (B, 256)
+        return self.head(x).squeeze(-1)
+
+
+def load_data(csv_path: str | Path) -> tuple[np.ndarray, np.ndarray, list[str]]:
+    _EXCLUDE = {"file_path", "label_int", "duration_sec", "sample_rate"}
+    rows, labels = [], []
+    with open(csv_path, "r", encoding="utf-8") as f:
+        reader = csv.DictReader(f)
+        feature_cols = [c for c in reader.fieldnames if c not in _EXCLUDE]
+        for row in reader:
+            vals = []
+            for col in feature_cols:
+                try:
+                    vals.append(float(row[col]))
+                except (ValueError, KeyError):
+                    vals.append(0.0)
+            rows.append(vals)
+            labels.append(int(row["label_int"]))
+    X = np.nan_to_num(np.array(rows, dtype=np.float32), nan=0.0)
+    y = np.array(labels, dtype=np.int32)
+    return X, y, feature_cols
+
+
+def train_one_fold(
+    model: nn.Module,
+    X_train: np.ndarray, y_train: np.ndarray,
+    X_val: np.ndarray, y_val: np.ndarray,
+) -> tuple[float, np.ndarray]:
+    scaler = StandardScaler()
+    X_tr = scaler.fit_transform(X_train)
+    X_v = scaler.transform(X_val)
+
+    train_ds = TensorDataset(
+        torch.tensor(X_tr, dtype=torch.float32),
+        torch.tensor(y_train, dtype=torch.float32),
+    )
+    val_X = torch.tensor(X_v, dtype=torch.float32).to(DEVICE)
+    val_y = torch.tensor(y_val, dtype=torch.float32)
+
+    loader = DataLoader(train_ds, batch_size=BATCH_SIZE, shuffle=True)
+    model = model.to(DEVICE)
+    optimizer = torch.optim.AdamW(model.parameters(), lr=LR, weight_decay=1e-4)
+    scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(
+        optimizer, mode="max", factor=0.5, patience=5
+    )
+    criterion = nn.BCEWithLogitsLoss()
+
+    best_auc = 0.0
+    best_probs = None
+    patience_ctr = 0
+
+    for epoch in range(EPOCHS):
+        model.train()
+        for bx, by in loader:
+            bx, by = bx.to(DEVICE), by.to(DEVICE)
+            optimizer.zero_grad()
+            logits = model(bx)
+            loss = criterion(logits, by)
+            loss.backward()
+            optimizer.step()
+
+        model.eval()
+        with torch.no_grad():
+            v_logits = model(val_X)
+            v_probs = torch.sigmoid(v_logits).cpu().numpy()
+
+        auc = roc_auc_score(y_val, v_probs)
+        scheduler.step(auc)
+
+        if auc > best_auc:
+            best_auc = auc
+            best_probs = v_probs.copy()
+            patience_ctr = 0
+        else:
+            patience_ctr += 1
+            if patience_ctr >= PATIENCE:
+                break
+
+    return best_auc, best_probs
+
+
+def evaluate_cv(
+    model_class: type,
+    X: np.ndarray, y: np.ndarray,
+    n_features: int,
+) -> dict:
+    cv = StratifiedKFold(n_splits=N_FOLDS, shuffle=True, random_state=SEED)
+    all_probs = np.zeros(len(y))
+    aucs = []
+    t0 = time.time()
+
+    for fold, (train_idx, val_idx) in enumerate(cv.split(X, y)):
+        set_seed(SEED + fold)
+        model = model_class(n_features)
+        auc, probs = train_one_fold(
+            model,
+            X[train_idx], y[train_idx],
+            X[val_idx], y[val_idx],
+        )
+        all_probs[val_idx] = probs
+        aucs.append(auc)
+        print(f"    Fold {fold+1}: AUC={auc:.4f}")
+
+    elapsed = time.time() - t0
+    y_pred = (all_probs > 0.5).astype(int)
+    return {
+        "accuracy": round(float(accuracy_score(y, y_pred)), 4),
+        "precision": round(float(precision_score(y, y_pred, zero_division=0)), 4),
+        "recall": round(float(recall_score(y, y_pred, zero_division=0)), 4),
+        "f1": round(float(f1_score(y, y_pred, zero_division=0)), 4),
+        "roc_auc": round(float(roc_auc_score(y, all_probs)), 4),
+        "fold_aucs": [round(a, 4) for a in aucs],
+        "train_time_sec": round(elapsed, 1),
+    }
+
+
+def main() -> None:
+    csv_path = sys.argv[1] if len(sys.argv) > 1 else "../DataSet/features.csv"
+    print(f"Device: {DEVICE}")
+    print(f"Loading: {csv_path}")
+
+    X, y, feature_cols = load_data(csv_path)
+    n_features = X.shape[1]
+    print(f"Samples: {len(y)}, Features: {n_features}")
+    print(f"AI: {np.sum(y == 1)}, Human: {np.sum(y == 0)}")
+
+    models = {
+        "Deep MLP (512-256-128-64)": DeepMLP,
+        "1D-CNN": Conv1DClassifier,
+        "Residual MLP (3 blocks)": ResidualMLP,
+        "Attention MLP": AttentionMLP,
+    }
+
+    all_results = {}
+    for name, cls in models.items():
+        print(f"\n{'='*60}")
+        print(f"  {name}")
+        print(f"{'='*60}")
+        result = evaluate_cv(cls, X, y, n_features)
+        all_results[name] = result
+        print(f"  => Acc={result['accuracy']:.4f}  AUC={result['roc_auc']:.4f}  "
+              f"F1={result['f1']:.4f}  Time={result['train_time_sec']:.0f}s")
+
+    out_path = Path("models/deep_learning_results.json")
+    with open(out_path, "w") as f:
+        json.dump(all_results, f, indent=2)
+    print(f"\nResults saved: {out_path}")
+
+    print(f"\n{'='*60}")
+    print("  SUMMARY")
+    print(f"{'='*60}")
+    for name, r in sorted(all_results.items(), key=lambda x: -x[1]["roc_auc"]):
+        print(f"  {name:35s} AUC={r['roc_auc']:.4f}  Acc={r['accuracy']:.4f}")
+
+
+if __name__ == "__main__":
+    main()