Skip to content

alinemol.utils

Helper functions for metrics, plotting, featurization, and model training used throughout the ID/OOD evaluation pipeline.

Optional extras

alinemol.utils re-exports symbols from several submodules. The plotting helpers are always available, but the metric, split, and training helpers depend on optional extras (torch/DGL from [gnn], statsmodels/astartes from [ml]). On a lean install, import the always-available symbols from alinemol.utils and the rest from their submodule directly, e.g. from alinemol.utils.utils import load_model.

Plotting

plot_ID_OOD

plot_ID_OOD(
    ID_test_score: List,
    OOD_test_score: List,
    threshold: float = 0.0,
    dataset_category: str = "MoleculeNet",
    dataset_name: str = "HIV",
    metric: str = "ROC-AUC",
    save: bool = False,
)

Plot ID vs OOD test ROC-AUC scores

Parameters:

Name Type Description Default
ID_test_score List

list of ID test scores

required
OOD_test_score List

list of OOD test scores

required
dataset_category str

category of dataset

'MoleculeNet'
dataset_name str

name of dataset

'HIV'
metric str

name of metric options: "ROC-AUC", "PR-AUC", "Accuracy"

'ROC-AUC'
save bool

whether to save plot

False

Returns:

Type Description

None

plot_ID_OOD_sns

plot_ID_OOD_sns(
    data: DataFrame,
    dataset_category="TDC",
    dataset_name="CYP2C19",
    save: bool = False,
)

Plot ID vs OOD test ROC-AUC scores using seaborn

Parameters:

Name Type Description Default
data DataFrame

DataFrame with the following columns: "model", "ID_test_accuracy", "OOD_test_accuracy", "ID_test_roc_auc", "OOD_test_roc_auc", "ID_test_pr_auc", "OOD_test_pr_auc"

required
dataset_category str

category of dataset

'TDC'
dataset_name str

name of dataset

'CYP2C19'
save bool

whether to save plot

False

Returns:

Type Description

None

visualize_chemspace

visualize_chemspace(
    data: DataFrame,
    split_names: List[str],
    mol_col: str = "smiles",
    size_col=None,
    size=10,
)

Visualize chemical space using UMAP

Parameters:

Name Type Description Default
data DataFrame

pd.DataFrame with columns "smiles", "label", "split"

required
split_names list

list of split names

required
mol_col str

name of column containing SMILES

'smiles'
size_col

name of column containing size information

None

Returns:

Type Description

None

Metrics

eval_roc_auc

eval_roc_auc(df1: DataFrame, df2: DataFrame) -> float

Evaluate ROC AUC score.

Parameters:

Name Type Description Default
df1 DataFrame

dataframe containing true labels.

required
df2 DataFrame

dataframe containing predicted labels.

required

Returns:

Name Type Description
float float

ROC AUC score.

eval_pr_auc

eval_pr_auc(df1: DataFrame, df2: DataFrame) -> float

Evaluate PR AUC score.

Parameters:

Name Type Description Default
df1 DataFrame

dataframe containing true labels.

required
df2 DataFrame

dataframe containing predicted labels.

required

Returns:

Name Type Description
float float

PR AUC score.

eval_acc

eval_acc(df1: DataFrame, df2: DataFrame, threshold: float = 0.5) -> float

Evaluate accuracy score.

Parameters:

Name Type Description Default
df1 DataFrame

dataframe containing true labels.

required
df2 DataFrame

dataframe containing predicted labels.

required
threshold float

threshold for binary classification.

0.5

Returns:

Name Type Description
float float

accuracy score.

compute_linear_fit

compute_linear_fit(
    x: Union[List[float], ndarray], y: Union[List[float], ndarray]
) -> Tuple[ndarray, float]

Returns bias and slope from regression y on x.

Parameters:

Name Type Description Default
x (list, ndarray)

x values.

required
y (list, ndarray)

y values.

required

Returns:

Name Type Description
tuple Tuple[ndarray, float]

first element is parameters and second element is rsquared.

compare_rankings

compare_rankings(
    condition1_values: Union[List[float], ndarray],
    condition2_values: Union[List[float], ndarray],
    category_names: Optional[List[str]] = None,
) -> Dict[str, Union[float, DataFrame]]

Compare rankings between two conditions using multiple metrics.

Parameters:

Name Type Description Default
condition1_values Union[List[float], ndarray]

list or array of values from condition 1

required
condition2_values Union[List[float], ndarray]

list or array of values from condition 2

required
category_names Optional[List[str]]

list of category names (optional)

None

Returns:

Name Type Description
dict Dict[str, Union[float, DataFrame]]

Dictionary containing various ranking comparison metrics

rescale

rescale(
    data: Union[List[float], ndarray], scaling: Optional[str] = None
) -> ndarray

Rescale the data.

Parameters:

Name Type Description Default
data (list, ndarray)

data to be rescaled.

required
scaling str

scaling method. Options: 'probit', 'logit', 'linear'.

None

Returns:

Type Description
ndarray

np.ndarray: rescaled data.

Meter

Meter(mean: Optional[Tensor] = None, std: Optional[Tensor] = None)

Track and summarize model performance on a dataset for (multi-label) prediction.

When dealing with multitask learning, quite often we normalize the labels so they are roughly at a same scale. During the evaluation, we need to undo the normalization on the predicted labels. If mean and std are not None, we will undo the normalization.

Currently we support evaluation with 4 metrics:

  • pearson r2
  • mae
  • rmse
  • roc auc score

Parameters:

Name Type Description Default
mean Optional[Tensor]

torch.float32 tensor of shape (T) or None. Mean of existing training labels across tasks if not None. T for the number of tasks. Default to None and we assume no label normalization has been performed.

None
std Optional[Tensor]

torch.float32 tensor of shape (T) Std of existing training labels across tasks if not None. Default to None and we assume no label normalization has been performed.

None

Examples:

Below gives a demo for a fake evaluation epoch.

>>> import torch
>>> from dgllife.utils import Meter
>>> meter = Meter()
>>> # Simulate 10 fake mini-batches
>>> for batch_id in range(10):
>>>     batch_label = torch.randn(3, 3)
>>>     batch_pred = torch.randn(3, 3)
>>>     meter.update(batch_pred, batch_label)
>>> # Get MAE for all tasks
>>> print(meter.compute_metric('mae'))
[1.1325558423995972, 1.0543707609176636, 1.094650149345398]
>>> # Get MAE averaged over all tasks
>>> print(meter.compute_metric('mae', reduction='mean'))
1.0938589175542195
>>> # Get the sum of MAE over all tasks
>>> print(meter.compute_metric('mae', reduction='sum'))
3.2815767526626587

update

update(y_pred: Tensor, y_true: Tensor, mask: Optional[Tensor] = None) -> None

Update for the result of an iteration

Parameters:

Name Type Description Default
y_pred Tensor

float32 tensor Predicted labels with shape (B, T), B for number of graphs in the batch and T for the number of tasks

required
y_true Tensor

float32 tensor Ground truth labels with shape (B, T)

required
mask Optional[Tensor]

None or float32 tensor Binary mask indicating the existence of ground truth labels with shape (B, T). If None, we assume that all labels exist and create a one-tensor for placeholder.

None

multilabel_score

multilabel_score(
    score_func: Callable[[Tensor, Tensor], float],
    reduction: ReductionType = "none",
) -> Union[float, List[float]]

Evaluate for multi-label prediction.

Parameters:

Name Type Description Default
score_func Callable[[Tensor, Tensor], float]

callable A score function that takes task-specific ground truth and predicted labels as input and return a float as the score. The labels are in the form of 1D tensor.

required
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

pearson_r2

pearson_r2(reduction: ReductionType = 'none') -> Union[float, List[float]]

Compute squared Pearson correlation coefficient.

Parameters:

Name Type Description Default
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

mae

mae(reduction: ReductionType = 'none') -> Union[float, List[float]]

Compute mean absolute error.

Parameters:

Name Type Description Default
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

rmse

rmse(reduction: ReductionType = 'none') -> Union[float, List[float]]

Compute root mean square error.

Parameters:

Name Type Description Default
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

accuracy_score

accuracy_score(
    reduction: ReductionType = "none", threshold: float = 0.5
) -> Union[float, List[float]]

Compute the accuracy score for binary classification. Accuracy scores are not well-defined in cases where labels for a task have one single class only (e.g. positive labels only or negative labels only). In this case we will simply ignore this task and print a warning message.

Parameters:

Name Type Description Default
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks.

'none'
threshold float

threshold for binary classification.

0.5

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

roc_auc_score

roc_auc_score(reduction: ReductionType = 'none') -> Union[float, List[float]]

Compute the area under the receiver operating characteristic curve (roc-auc score) for binary classification.

ROC-AUC scores are not well-defined in cases where labels for a task have one single class only (e.g. positive labels only or negative labels only). In this case we will simply ignore this task and print a warning message.

Parameters:

Name Type Description Default
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks.

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

pr_auc_score

pr_auc_score(reduction: ReductionType = 'none') -> Union[float, List[float]]

Compute the area under the precision-recall curve (pr-auc score) for binary classification.

PR-AUC scores are not well-defined in cases where labels for a task have one single class only (e.g. positive labels only or negative labels only). In this case, we will simply ignore this task and print a warning message.

Parameters:

Name Type Description Default
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks.

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

compute_metric

compute_metric(
    metric_name: str, reduction: ReductionType = "none"
) -> Union[float, List[float]]

Compute metric based on metric name.

Parameters:

Name Type Description Default
metric_name str

str

  • 'r2': compute squared Pearson correlation coefficient
  • 'mae': compute mean absolute error
  • 'rmse': compute root mean square error
  • 'roc_auc_score': compute roc-auc score
  • 'pr_auc_score': compute pr-auc score
required
reduction ReductionType

'none' or 'mean' or 'sum' Controls the form of scores for all tasks

'none'

Returns:

Type Description
Union[float, List[float]]

float or list of float * If reduction == 'none', return the list of scores for all tasks. * If reduction == 'mean', return the mean of scores for all tasks. * If reduction == 'sum', return the sum of scores for all tasks.

Splitting helpers

compute_similarities

compute_similarities(
    source_molecules: Union[List, ndarray],
    target_molecules: Union[List, ndarray],
    fingerprint: str,
    fprints_hopts: Dict,
) -> ndarray

Compute similarities between two lists of molecules. It receives two lists of smiles or RDKit molecule objects, extracts their fingerprints and computes the similarities between them.

Parameters:

Name Type Description Default
source_molecules array or list

SMILES strings or RDKit molecule objects.

required
target_molecules array or list

SMILES strings or RDKit molecule objects.

required
fingerprint str

The molecular fingerprint to be used.

required
fprints_hopts dict

Hyperparameters for AIMSim.

required

Returns:

Type Description
ndarray

np.ndarray: Matrix of similarities between the two lists of molecules

featurize

featurize(
    molecules: Union[List, ndarray], fingerprint: str, fprints_hopts: Dict
) -> ndarray

Call AIMSim's Molecule to featurize the molecules according to the arguments.

Parameters:

Name Type Description Default
molecules array or list

SMILES strings or RDKit molecule objects.

required
fingerprint str

The molecular fingerprint to be used.

required
fprints_hopts dict

Hyperparameters for AIMSim.

required

Returns:

Type Description
ndarray

np.ndarray: X array (featurized molecules)

Raises:

Type Description
ImportError

If aimsim is not installed.

split_molecules_train_test

split_molecules_train_test(
    mol_df: DataFrame,
    sampler: str,
    train_size: float = 0.9,
    random_state: int = 42,
    hopts: dict = {},
) -> Tuple[DataFrame, DataFrame]

Split molecules into train and test sets.

Parameters:

Name Type Description Default
mol_df DataFrame

Dataframe of molecules. It must have two columns: 'smiles' and 'label'.

required
sampler str

Sampler to use. Options: random, scaffold, kmeans, dbscan, sphere_exclusion, optisim.

required
train_size float

Size of the train set.

0.9
random_state int

Random state for reproducibility.

42
hopts dict

Hyperparameters for the sampler.

{}

Returns:

Name Type Description
tuple Tuple[DataFrame, DataFrame]

Tuple containing the train and test sets.

split_molecules_train_val_test

split_molecules_train_val_test(
    mol_df: DataFrame,
    sampler: str,
    train_size: float = 0.8,
    val_size: float = 0.1,
    random_state: int = 42,
    hopts: dict = {},
) -> Tuple[DataFrame, DataFrame, DataFrame]

Split molecules into train and test sets.

Parameters:

Name Type Description Default
mol_df DataFrame

Dataframe of molecules. It must have two columns: 'smiles' and 'label'.

required
sampler str

Sampler to use. Options: RandomSplit, ScaffoldSplit, KMeansSplit, DBScanSplit, SphereExclusionSplit, OptiSimSplit.

required
train_size float

Size of the train set.

0.8
val_size float

Size of the validation set.

0.1
random_state int

Random state for reproducibility.

42
hopts dict

Hyperparameters for the sampler.

{}

Returns:

Name Type Description
tuple Tuple[DataFrame, DataFrame, DataFrame]

Tuple containing the train and test sets.

Training & data loading

load_dataset

load_dataset(args: ConfigDict, df: DataFrame) -> MoleculeCSVDataset

Load the dataset

Parameters:

Name Type Description Default
args dict

Settings

required
df DataFrame

Dataframe

required

Returns:

Type Description
MoleculeCSVDataset

MoleculeCSVDataset

Raises:

Type Description
ValueError

If args does not contain the key 'smiles_column', 'task_names', 'result_path', 'num_workers'

Examples:

>>> from alinemol.utils.utils import load_dataset
>>> args = {
...    "smiles_column": "smiles",
...    "task_names": "task_1,task_2",
...    "result_path": "results",
...    "num_workers": 4,
... }
>>> df = pd.read_csv("data.csv")
>>> dataset = load_dataset(args, df)
>>> print(dataset)

load_model

load_model(exp_configure: ConfigDict) -> ModelType

Build a model from an experiment configuration.

Parameters:

Name Type Description Default
exp_configure dict

Experiment configuration mapping (model name and its hyperparameters).

required

Returns:

Type Description
ModelType

dgllife.model: The instantiated model.

init_featurizer

init_featurizer(args: ConfigDict) -> ConfigDict

Initialize node/edge featurizer

Parameters:

Name Type Description Default
args dict

Settings

required

Returns:

Name Type Description
dict ConfigDict

Settings with featurizers updated

Raises:

Type Description
ValueError

If the node_featurizer_type is not in ['canonical', 'attentivefp']

ValueError

If args does not contain the key 'model', 'atom_featurizer_type', 'bond_featurizer_type'

Examples:

>>> from alinemol.utils.utils import init_featurizer
>>> args = {
...    "model": "GCN",
...    "atom_featurizer_type": "canonical",
...    "bond_featurizer_type": "canonical",
... }
>>> args = init_featurizer(args)
>>> print(args)

get_configure

get_configure(model: str) -> ConfigDict

Query for the manually specified configuration

Parameters:

Name Type Description Default
model str

Model type

required

Returns:

Name Type Description
dict ConfigDict

Returns the manually specified configuration

collate_molgraphs

collate_molgraphs(
    data: List[Tuple[str, DGLGraph, Tensor, Tensor]],
) -> Tuple[List[str], DGLGraph, Tensor, Tensor]

Batching a list of datapoints for dataloader.

Parameters:

Name Type Description Default
data list of 4-tuples

Each tuple is for a single datapoint, consisting of a SMILES, a DGLGraph, all-task labels and a binary mask indicating the existence of labels.

required

Returns:

Name Type Description
smiles list

List of smiles

bg DGLGraph

The batched DGLGraph.

labels Tensor

Tensor of dtype float32 and shape (B, T) Batched datapoint labels. B is len(data) and T is the number of total tasks.

masks Tensor

Tensor of dtype float32 and shape (B, T) Batched datapoint binary mask, indicating the existence of labels.

predict

predict(args: ConfigDict, model: ModelType, bg: DGLGraph) -> Tensor

Predict the output of the models for the input batch graphs.

Parameters:

Name Type Description Default
args dict

Runtime configuration (e.g. device and node/edge feature keys).

required
model Module

The model to predict

required
bg DGLGraph

The input batch graphs

required

Returns:

Type Description
Tensor

Torch.Tesnor

split_dataset

split_dataset(
    args: ConfigDict, dataset: DatasetType
) -> Tuple[DatasetType, DatasetType, DatasetType]

Split the dataset into train, validation and test sets.

Parameters:

Name Type Description Default
args dict

Settings containing split method and ratios

required
dataset MoleculeCSVDataset

Dataset to split

required

Returns:

Name Type Description
tuple Tuple[DatasetType, DatasetType, DatasetType]

(train_dataset, val_dataset, test_dataset)

Raises:

Type Description
ValueError

If split method is invalid or split ratios are invalid

Examples:

>>> from alinemol.utils.utils import split_dataset
>>> args = {
...    "split": "scaffold_decompose",
...    "split_ratio": "0.8,0.1,0.1",
... }
>>> train_set, val_set, test_set = split_dataset(args, dataset)
>>> print(train_set)