geneformer/mtl/utils.py

import os
import shutil

from sklearn.metrics import accuracy_score, f1_score
from sklearn.preprocessing import LabelEncoder
from transformers import AutoConfig, BertConfig, BertModel

from .imports import *


def save_model(model, model_save_directory):
    if not os.path.exists(model_save_directory):
        os.makedirs(model_save_directory)

    # Get the state dict
    if isinstance(model, nn.DataParallel):
        model_state_dict = (
            model.module.state_dict()
        )  # Use model.module to access the underlying model
    else:
        model_state_dict = model.state_dict()

    # Remove the "module." prefix from the keys if present
    model_state_dict = {
        k.replace("module.", ""): v for k, v in model_state_dict.items()
    }

    model_save_path = os.path.join(model_save_directory, "pytorch_model.bin")
    torch.save(model_state_dict, model_save_path)

    # Save the model configuration
    if isinstance(model, nn.DataParallel):
        model.module.config.to_json_file(
            os.path.join(model_save_directory, "config.json")
        )
    else:
        model.config.to_json_file(os.path.join(model_save_directory, "config.json"))

    print(f"Model and configuration saved to {model_save_directory}")


def calculate_task_specific_metrics(task_true_labels, task_pred_labels):
    task_metrics = {}
    for task_name in task_true_labels.keys():
        true_labels = task_true_labels[task_name]
        pred_labels = task_pred_labels[task_name]
        f1 = f1_score(true_labels, pred_labels, average="macro")
        accuracy = accuracy_score(true_labels, pred_labels)
        task_metrics[task_name] = {"f1": f1, "accuracy": accuracy}
    return task_metrics


def calculate_combined_f1(combined_labels, combined_preds):
    # Initialize the LabelEncoder
    le = LabelEncoder()

    # Fit and transform combined labels and predictions to numerical values
    le.fit(combined_labels + combined_preds)
    encoded_true_labels = le.transform(combined_labels)
    encoded_pred_labels = le.transform(combined_preds)

    # Print out the mapping for sanity check
    print("\nLabel Encoder Mapping:")
    for index, class_label in enumerate(le.classes_):
        print(f"'{class_label}': {index}")

    # Calculate accuracy
    accuracy = accuracy_score(encoded_true_labels, encoded_pred_labels)

    # Calculate F1 Macro score
    f1 = f1_score(encoded_true_labels, encoded_pred_labels, average="macro")

    return f1, accuracy


def save_model_without_heads(original_model_save_directory):
    # Create a new directory for the model without heads
    new_model_save_directory = original_model_save_directory + "_No_Heads"
    if not os.path.exists(new_model_save_directory):
        os.makedirs(new_model_save_directory)

    # Load the model state dictionary
    model_state_dict = torch.load(
        os.path.join(original_model_save_directory, "pytorch_model.bin")
    )

    # Initialize a new BERT model without the classification heads
    config = BertConfig.from_pretrained(
        os.path.join(original_model_save_directory, "config.json")
    )
    model_without_heads = BertModel(config)

    # Filter the state dict to exclude classification heads
    model_without_heads_state_dict = {
        k: v
        for k, v in model_state_dict.items()
        if not k.startswith("classification_heads")
    }

    # Load the filtered state dict into the model
    model_without_heads.load_state_dict(model_without_heads_state_dict, strict=False)

    # Save the model without heads
    model_save_path = os.path.join(new_model_save_directory, "pytorch_model.bin")
    torch.save(model_without_heads.state_dict(), model_save_path)

    # Copy the configuration file
    shutil.copy(
        os.path.join(original_model_save_directory, "config.json"),
        new_model_save_directory,
    )

    print(f"Model without classification heads saved to {new_model_save_directory}")


def get_layer_freeze_range(pretrained_path):
    """
    Dynamically determines the number of layers to freeze based on the model depth from its configuration.
    Args:
        pretrained_path (str): Path to the pretrained model directory or model identifier.
    Returns:
        dict: A dictionary with 'min' and 'max' keys indicating the range of layers to freeze.
    """
    if pretrained_path:
        config = AutoConfig.from_pretrained(pretrained_path)
        total_layers = config.num_hidden_layers
        return {"min": 0, "max": total_layers - 1}
    else:
        return {"min": 0, "max": 0}