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
¶
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
¶
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
¶
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 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
¶
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 r2maermseroc 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
|
std
|
Optional[Tensor]
|
torch.float32 tensor of shape (T)
Std of existing training labels across tasks if not |
None
|
Examples:
Below gives a demo for a fake evaluation epoch.
>>> 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 for the result of an iteration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
y_pred
|
Tensor
|
float32 tensor
Predicted labels with shape |
required |
y_true
|
Tensor
|
float32 tensor
Ground truth labels with shape |
required |
mask
|
Optional[Tensor]
|
None or float32 tensor
Binary mask indicating the existence of ground truth labels with
shape |
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 |
pearson_r2
¶
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 |
mae
¶
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 |
rmse
¶
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 |
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 |
roc_auc_score
¶
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 |
pr_auc_score
¶
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 |
compute_metric
¶
Compute metric based on metric name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
metric_name
|
str
|
str
|
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 |
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
¶
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 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
¶
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
¶
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:
get_configure
¶
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 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: