geneformer/mtl/train.py

from .imports import *
from .data import preload_and_process_data, get_data_loader
from .model import GeneformerMultiTask
from .utils import calculate_task_specific_metrics
from torch.utils.tensorboard import SummaryWriter
import pandas as pd
import os
from tqdm import tqdm
import random
import numpy as np
import torch


def set_seed(seed):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

def initialize_wandb(config):
    if config.get("use_wandb", False):
        import wandb
        wandb.init(project=config["wandb_project"], config=config)
        print("Weights & Biases (wandb) initialized and will be used for logging.")
    else:
        print("Weights & Biases (wandb) is not enabled. Logging will use other methods.")

def create_model(config, num_labels_list, device):
    model = GeneformerMultiTask(
        config["pretrained_path"],
        num_labels_list,
        dropout_rate=config["dropout_rate"],
        use_task_weights=config["use_task_weights"],
        task_weights=config["task_weights"],
        max_layers_to_freeze=config["max_layers_to_freeze"],
        use_attention_pooling=config["use_attention_pooling"]
    )
    if config["use_data_parallel"]:
        model = nn.DataParallel(model)
    return model.to(device)

def setup_optimizer_and_scheduler(model, config, total_steps):
    optimizer = AdamW(model.parameters(), lr=config["learning_rate"], weight_decay=config["weight_decay"])
    warmup_steps = int(config["warmup_ratio"] * total_steps)

    if config["lr_scheduler_type"] == "linear":
        scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=warmup_steps, num_training_steps=total_steps)
    elif config["lr_scheduler_type"] == "cosine":
        scheduler = get_cosine_schedule_with_warmup(optimizer, num_warmup_steps=warmup_steps, num_training_steps=total_steps, num_cycles=0.5)

    return optimizer, scheduler

def train_epoch(model, train_loader, optimizer, scheduler, device, config, writer, epoch):
    model.train()
    progress_bar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{config['epochs']}")
    for batch_idx, batch in enumerate(progress_bar):
        optimizer.zero_grad()
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = [batch['labels'][task_name].to(device) for task_name in config["task_names"]]

        loss, _, _ = model(input_ids, attention_mask, labels)
        loss.backward()

        if config["gradient_clipping"]:
            torch.nn.utils.clip_grad_norm_(model.parameters(), config["max_grad_norm"])

        optimizer.step()
        scheduler.step()

        writer.add_scalar('Training Loss', loss.item(), epoch * len(train_loader) + batch_idx)
        if config.get("use_wandb", False):
            wandb.log({'Training Loss': loss.item()})

        # Update progress bar
        progress_bar.set_postfix({'loss': f"{loss.item():.4f}"})

    return loss.item()  # Return the last batch loss

def validate_model(model, val_loader, device, config):
    model.eval()
    val_loss = 0.0
    task_true_labels = {task_name: [] for task_name in config["task_names"]}
    task_pred_labels = {task_name: [] for task_name in config["task_names"]}
    task_pred_probs = {task_name: [] for task_name in config["task_names"]}

    with torch.no_grad():
        for batch in val_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = [batch['labels'][task_name].to(device) for task_name in config["task_names"]]
            loss, logits, _ = model(input_ids, attention_mask, labels)
            val_loss += loss.item()

            for sample_idx in range(len(batch['input_ids'])):
                for i, task_name in enumerate(config["task_names"]):
                    true_label = batch['labels'][task_name][sample_idx].item()
                    pred_label = torch.argmax(logits[i][sample_idx], dim=-1).item()
                    pred_prob = torch.softmax(logits[i][sample_idx], dim=-1).cpu().numpy()
                    task_true_labels[task_name].append(true_label)
                    task_pred_labels[task_name].append(pred_label)
                    task_pred_probs[task_name].append(pred_prob)

    val_loss /= len(val_loader)
    return val_loss, task_true_labels, task_pred_labels, task_pred_probs

def log_metrics(task_metrics, val_loss, config, writer, epochs):
    for task_name, metrics in task_metrics.items():
        print(f"{task_name} - Validation F1 Macro: {metrics['f1']:.4f}, Validation Accuracy: {metrics['accuracy']:.4f}")
        if config.get("use_wandb", False):
            import wandb
            wandb.log({
                f'{task_name} Validation F1 Macro': metrics['f1'],
                f'{task_name} Validation Accuracy': metrics['accuracy']
            })

    writer.add_scalar('Validation Loss', val_loss, epochs)
    for task_name, metrics in task_metrics.items():
        writer.add_scalar(f'{task_name} - Validation F1 Macro', metrics['f1'], epochs)
        writer.add_scalar(f'{task_name} - Validation Accuracy', metrics['accuracy'], epochs)

def save_validation_predictions(val_cell_id_mapping, task_true_labels, task_pred_labels, task_pred_probs, config, trial_number=None):
    if trial_number is not None:
        trial_results_dir = os.path.join(config["results_dir"], f"trial_{trial_number}")
        os.makedirs(trial_results_dir, exist_ok=True)
        val_preds_file = os.path.join(trial_results_dir, "val_preds.csv")
    else:
        val_preds_file = os.path.join(config["results_dir"], "manual_run_val_preds.csv")
    
    rows = []
    for sample_idx in range(len(val_cell_id_mapping)):
        row = {'Cell ID': val_cell_id_mapping[sample_idx]}
        for task_name in config["task_names"]:
            row[f'{task_name} True'] = task_true_labels[task_name][sample_idx]
            row[f'{task_name} Pred'] = task_pred_labels[task_name][sample_idx]
            row[f'{task_name} Probabilities'] = ','.join(map(str, task_pred_probs[task_name][sample_idx]))
        rows.append(row)
    
    df = pd.DataFrame(rows)
    df.to_csv(val_preds_file, index=False)
    print(f"Validation predictions saved to {val_preds_file}")


def train_model(config, device, train_loader, val_loader, train_cell_id_mapping, val_cell_id_mapping, num_labels_list):
    set_seed(config["seed"])
    initialize_wandb(config)

    model = create_model(config, num_labels_list, device)
    total_steps = len(train_loader) * config["epochs"]
    optimizer, scheduler = setup_optimizer_and_scheduler(model, config, total_steps)

    log_dir = os.path.join(config["tensorboard_log_dir"], "manual_run")
    writer = SummaryWriter(log_dir=log_dir)

    epoch_progress = tqdm(range(config["epochs"]), desc="Training Progress")
    for epoch in epoch_progress:
        last_loss = train_epoch(model, train_loader, optimizer, scheduler, device, config, writer, epoch)
        epoch_progress.set_postfix({'last_loss': f"{last_loss:.4f}"})

    val_loss, task_true_labels, task_pred_labels, task_pred_probs = validate_model(model, val_loader, device, config)
    task_metrics = calculate_task_specific_metrics(task_true_labels, task_pred_labels)
    
    log_metrics(task_metrics, val_loss, config, writer, config["epochs"])
    writer.close()

    save_validation_predictions(val_cell_id_mapping, task_true_labels, task_pred_labels, task_pred_probs, config)

    if config.get("use_wandb", False):
        import wandb
        wandb.finish()

    print(f"\nFinal Validation Loss: {val_loss:.4f}")
    return val_loss, model  # Return both the validation loss and the trained model

def objective(trial, train_loader, val_loader, train_cell_id_mapping, val_cell_id_mapping, num_labels_list, config, device):
    set_seed(config["seed"])  # Set the seed before each trial
    initialize_wandb(config)

    # Hyperparameters
    config["learning_rate"] = trial.suggest_float("learning_rate", config["hyperparameters"]["learning_rate"]["low"], config["hyperparameters"]["learning_rate"]["high"], log=config["hyperparameters"]["learning_rate"]["log"])
    config["warmup_ratio"] = trial.suggest_float("warmup_ratio", config["hyperparameters"]["warmup_ratio"]["low"], config["hyperparameters"]["warmup_ratio"]["high"])
    config["weight_decay"] = trial.suggest_float("weight_decay", config["hyperparameters"]["weight_decay"]["low"], config["hyperparameters"]["weight_decay"]["high"])
    config["dropout_rate"] = trial.suggest_float("dropout_rate", config["hyperparameters"]["dropout_rate"]["low"], config["hyperparameters"]["dropout_rate"]["high"])
    config["lr_scheduler_type"] = trial.suggest_categorical("lr_scheduler_type", config["hyperparameters"]["lr_scheduler_type"]["choices"])
    config["use_attention_pooling"] = trial.suggest_categorical("use_attention_pooling", [True, False])

    if config["use_task_weights"]:
        config["task_weights"] = [trial.suggest_float(f"task_weight_{i}", config["hyperparameters"]["task_weights"]["low"], config["hyperparameters"]["task_weights"]["high"]) for i in range(len(num_labels_list))]
        weight_sum = sum(config["task_weights"])
        config["task_weights"] = [weight / weight_sum for weight in config["task_weights"]]
    else:
        config["task_weights"] = None

    # Fix for max_layers_to_freeze
    if isinstance(config["max_layers_to_freeze"], dict):
        config["max_layers_to_freeze"] = trial.suggest_int("max_layers_to_freeze", config["max_layers_to_freeze"]["min"], config["max_layers_to_freeze"]["max"])
    elif isinstance(config["max_layers_to_freeze"], int):
        # If it's already an int, we don't need to suggest it
        pass
    else:
        raise ValueError("Invalid type for max_layers_to_freeze. Expected dict or int.")

    model = create_model(config, num_labels_list, device)
    total_steps = len(train_loader) * config["epochs"]
    optimizer, scheduler = setup_optimizer_and_scheduler(model, config, total_steps)

    log_dir = os.path.join(config["tensorboard_log_dir"], f"trial_{trial.number}")
    writer = SummaryWriter(log_dir=log_dir)

    for epoch in range(config["epochs"]):
        train_epoch(model, train_loader, optimizer, scheduler, device, config, writer, epoch)

    val_loss, task_true_labels, task_pred_labels, task_pred_probs = validate_model(model, val_loader, device, config)
    task_metrics = calculate_task_specific_metrics(task_true_labels, task_pred_labels)
    
    log_metrics(task_metrics, val_loss, config, writer, config["epochs"])
    writer.close()

    save_validation_predictions(val_cell_id_mapping, task_true_labels, task_pred_labels, task_pred_probs, config, trial.number)

    trial.set_user_attr("model_state_dict", model.state_dict())
    trial.set_user_attr("task_weights", config["task_weights"])

    trial.report(val_loss, config["epochs"])

    if trial.should_prune():
        raise optuna.TrialPruned()

    if config.get("use_wandb", False):
        import wandb
        wandb.log({
            "trial_number": trial.number,
            "val_loss": val_loss,
            **{f"{task_name}_f1": metrics['f1'] for task_name, metrics in task_metrics.items()},
            **{f"{task_name}_accuracy": metrics['accuracy'] for task_name, metrics in task_metrics.items()},
            **{k: v for k, v in config.items() if k in ["learning_rate", "warmup_ratio", "weight_decay", "dropout_rate", "lr_scheduler_type", "use_attention_pooling", "max_layers_to_freeze"]}
        })
        wandb.finish()

    return val_loss