"""Pooled-screen simulation, evaluation, and visualisation utilities."""
import os, random, warnings, traceback, sqlite3, math
from . import _gc as gc
from time import time, sleep
from datetime import datetime
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from matplotlib.patches import Patch
import seaborn as sns
import sklearn.metrics as metrics
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.inspection import PartialDependenceDisplay, permutation_importance
from sklearn.metrics import roc_curve, auc, confusion_matrix, precision_recall_curve
import statsmodels.api as sm
from multiprocessing import cpu_count, Manager
from .resource_log import _parallel_pool as Pool
from copy import deepcopy
from io import StringIO
from .figures.style import (ROLES, TYPE_SCALE, Palette, figure_style,
reference_line, resolve_ink, theme_target)
from .plot import save_figure
#: The simulator's own vocabulary, from the published palette. These are
#: DIAGNOSTICS of a simulation rather than a user's data, but the same rule
#: applies: the guides that were made active are the claim, everything else
#: is the ground.
SIM_INACTIVE = ROLES['data']
SIM_ACTIVE = ROLES['highlight']
SIM_MIXED = Palette.RUST
SIM_CONTROL = ROLES['control_negative']
warnings.filterwarnings("ignore")
warnings.filterwarnings("ignore", category=RuntimeWarning)
def _style_colour_bar(fig):
"""Put a seaborn heatmap's colour bar into the house style.
``sns.heatmap`` builds the bar's axes after the figure exists, with
matplotlib's default black ticks and a full frame -- a white box around
the ramp on the dark ground.
"""
if len(fig.axes) < 2:
return
ink = resolve_ink(theme_target())
bar = fig.axes[-1]
bar.tick_params(colors=ink, labelsize=TYPE_SCALE['tick'])
for spine in bar.spines.values():
spine.set_visible(False)
[docs]
def generate_gene_list(number_of_genes, number_of_all_genes):
"""Return ``number_of_genes`` randomly-drawn gene indices without replacement.
:param number_of_genes: Number of gene indices to draw.
:param number_of_all_genes: Size of the pool ``[0, number_of_all_genes)``.
:returns: List of drawn gene indices.
"""
genes_ls = list(range(number_of_all_genes))
random.shuffle(genes_ls)
gene_list = genes_ls[:number_of_genes]
return gene_list
[docs]
def generate_plate_map(nr_plates):
"""Return a 384-well plate map DataFrame spanning ``nr_plates`` plates.
:param nr_plates: Number of plates to enumerate.
:returns: DataFrame with ``plate_row_column``, ``plate_id``, ``row_id``,
``column_id`` columns (16 rows x 24 columns per plate).
"""
plate_row_column = [f"{i+1}_{ir+1}_{ic+1}" for i in range(nr_plates) for ir in range(16) for ic in range(24)]
df= pd.DataFrame({'plate_row_column': plate_row_column})
df["plate_id"], df["row_id"], df["column_id"] = zip(*[r.split("_") for r in df['plate_row_column']])
return df
[docs]
def gini_coefficient(x):
"""Return the Gini coefficient of ``x`` via the pairwise absolute difference formula.
:param x: 1-D array-like of non-negative values.
:returns: Gini coefficient in ``[0, 1]``.
"""
diffsum = np.sum(np.abs(np.subtract.outer(x, x)))
return diffsum / (2 * len(x) ** 2 * np.mean(x))
[docs]
def gini_gene_well(x):
"""Return the Gini coefficient of ``x`` using a memory-cheap upper-triangle sum.
:param x: 1-D array-like of non-negative values.
:returns: Gini coefficient in ``[0, 1]``; 0 is perfect equality, 1 is perfect inequality.
"""
total = 0
for i, xi in enumerate(x[:-1], 1):
total += np.sum(np.abs(xi - x[i:]))
return total / (len(x)**2 * np.mean(x))
[docs]
def gini(x):
"""Return the Gini coefficient of ``x`` via the ranked-sum formulation.
Reference: StatsDirect non-parametric methods
(https://www.statsdirect.com/help/nonparametric_methods/gini.htm).
:param x: 1-D array-like; all values treated equally.
:returns: Gini coefficient in ``[0, 1]``.
"""
x = np.array(x, dtype=np.float64)
n = len(x)
s = x.sum()
r = np.argsort(np.argsort(-x))
return 1 - (2 * (r * x).sum() + s) / (n * s)
[docs]
def dist_gen(mean, sd, df):
"""Draw a length-``len(df)`` Poisson sample with gamma-distributed rates.
:param mean: Mean of the gamma prior on the Poisson rate.
:param sd: Standard deviation of the gamma prior.
:param df: DataFrame whose length sets the sample size.
:returns: Tuple ``(samples, length)`` where ``samples`` is a NumPy array of
Poisson draws and ``length`` is ``len(df)``.
"""
length = len(df)
shape = (mean / sd) ** 2
scale = (sd ** 2) / mean
rate = np.random.gamma(shape, scale, size=length)
data = np.random.poisson(rate)
return data, length
[docs]
def generate_gene_weights(positive_mean, positive_variance, df):
"""Draw ``len(df)`` gene weights from a Beta distribution matched to the given moments.
:param positive_mean: Target mean of the Beta distribution in ``(0, 1)``.
:param positive_variance: Target variance (must be feasible for the mean).
:param df: DataFrame whose length sets the sample size.
:returns: NumPy array of Beta-distributed weights.
"""
a1 = positive_mean*(positive_mean*(1-positive_mean)/positive_variance - 1)
b1 = a1*(1-positive_mean)/positive_mean
weights = np.random.beta(a1, b1, len(df))
return weights
[docs]
def normalize_array(arr):
"""Return ``arr`` min-max scaled into ``[0, 1]``.
:param arr: Input NumPy array.
:returns: Normalized array of the same shape.
"""
min_value = np.min(arr)
max_value = np.max(arr)
value_range = max_value - min_value
if value_range == 0:
return np.zeros_like(arr, dtype=float)
return (arr - min_value) / value_range
[docs]
def generate_power_law_distribution(num_elements, coeff):
"""Return a normalized power-law probability vector of length ``num_elements``.
:param num_elements: Length of the returned distribution.
:param coeff: Positive exponent applied as ``i^-coeff``.
:returns: NumPy array that sums to 1.
"""
base_distribution = np.arange(1, num_elements + 1)
powered_distribution = base_distribution ** -coeff
normalized_distribution = powered_distribution / np.sum(powered_distribution)
return normalized_distribution
[docs]
def power_law_dist_gen(df, avg, well_ineq_coeff):
"""Return ``len(df)`` per-well values sampled from an average-scaled power-law.
:param df: DataFrame of wells whose length sets the sample size.
:param avg: Scale factor applied to each drawn probability.
:param well_ineq_coeff: Power-law exponent (larger = more unequal).
:returns: NumPy array of per-well quantities.
"""
distribution = generate_power_law_distribution(len(df), well_ineq_coeff)
dist = np.random.choice(distribution, len(df)) * avg
return dist
[docs]
def run_experiment(plate_map, number_of_genes, active_gene_list, avg_genes_per_well, sd_genes_per_well, avg_cells_per_well, sd_cells_per_well, well_ineq_coeff, gene_ineq_coeff):
"""Simulate one cell-level screening experiment and return per-cell + summary tables.
Draws per-well gene assignments from a power-law distribution and per-well
cell counts from a gamma/Poisson mixture, then labels each cell active/inactive.
:param plate_map: DataFrame of wells with plate/row/column identifiers.
:param number_of_genes: Total number of genes in the pool.
:param active_gene_list: Gene indices considered active (positive class).
:param avg_genes_per_well: Mean genes-per-well before power-law scaling.
:param sd_genes_per_well: Standard deviation of genes-per-well.
:param avg_cells_per_well: Mean cells-per-well.
:param sd_cells_per_well: Standard deviation of cells-per-well.
:param well_ineq_coeff: Power-law exponent for well-level inequality.
:param gene_ineq_coeff: Power-law exponent for gene-level inequality.
:returns: Tuple ``(cell_df, genes_per_well_df, wells_per_gene_df, df_ls)``
where ``df_ls`` contains per-well gene counts, per-gene well counts,
per-well Gini values, per-gene Gini values, gene weights and well weights.
"""
plate_map = plate_map.reset_index(drop=True)
cpw, _ = dist_gen(avg_cells_per_well, sd_cells_per_well, plate_map)
gpw, _ = dist_gen(avg_genes_per_well, sd_genes_per_well, plate_map)
genes = [*range(1, number_of_genes+1, 1)]
gene_weights = {gene: weight for gene, weight in zip(genes, generate_power_law_distribution(number_of_genes, gene_ineq_coeff))}
gene_weights_array = np.array(list(gene_weights.values()))
well_weights = generate_power_law_distribution(len(plate_map), well_ineq_coeff)
gene_to_well_mapping = {}
for gene in range(1, number_of_genes + 1):
if gene-1 < len(gpw):
max_index = len(plate_map['plate_row_column'])
num_samples = int(gpw[gene-1])
if num_samples >= max_index:
num_samples = max_index - 1
gene_to_well_mapping[gene] = np.random.choice(plate_map['plate_row_column'], size=num_samples, replace=False, p=well_weights)
else:
break
cells = []
for i in [*range(0,len(plate_map))]:
ciw = random.choice(cpw)
present_genes = [gene for gene, wells in gene_to_well_mapping.items() if plate_map.loc[i, 'plate_row_column'] in wells]
present_gene_weights = [gene_weights[gene] for gene in present_genes]
present_gene_weights /= np.sum(present_gene_weights)
if present_genes:
giw = np.random.choice(present_genes, int(gpw[i]), p=present_gene_weights)
if len(giw) > 0:
for _ in range(0,int(ciw)):
gene_nr = random.choice(giw)
cell = {
'plate_row_column': plate_map.loc[i, 'plate_row_column'],
'plate_id': plate_map.loc[i, 'plate_id'],
'row_id': plate_map.loc[i, 'row_id'],
'column_id': plate_map.loc[i, 'column_id'],
'genes_in_well': len(giw),
'gene_id': gene_nr,
'is_active': int(gene_nr in active_gene_list)
}
cells.append(cell)
cell_df = pd.DataFrame(cells)
cell_df = cell_df.dropna()
gene_counts_per_well = cell_df.groupby('plate_row_column')['gene_id'].nunique().sort_values().tolist()
well_counts_per_gene = cell_df.groupby('gene_id')['plate_row_column'].nunique().sort_values().tolist()
genes_per_well_df = pd.DataFrame(gene_counts_per_well, columns=['genes_per_well'])
genes_per_well_df['rank'] = range(1, len(genes_per_well_df) + 1)
wells_per_gene_df = pd.DataFrame(well_counts_per_gene, columns=['wells_per_gene'])
wells_per_gene_df['rank'] = range(1, len(wells_per_gene_df) + 1)
ls_ = []
gini_ls = []
for i,val in enumerate(cell_df['plate_row_column'].unique().tolist()):
temp = cell_df[cell_df['plate_row_column']==val]
x = temp['gene_id'].value_counts().to_numpy()
gini_val = gini_gene_well(x)
ls_.append(val)
gini_ls.append(gini_val)
gini_well = np.array(gini_ls)
ls_ = []
gini_ls = []
for i,val in enumerate(cell_df['gene_id'].unique().tolist()):
temp = cell_df[cell_df['gene_id']==val]
x = temp['plate_row_column'].value_counts().to_numpy()
gini_val = gini_gene_well(x)
ls_.append(val)
gini_ls.append(gini_val)
gini_gene = np.array(gini_ls)
df_ls = [gene_counts_per_well, well_counts_per_gene, gini_well, gini_gene, gene_weights_array, well_weights]
return cell_df, genes_per_well_df, wells_per_gene_df, df_ls
[docs]
def classifier(positive_mean, positive_variance, negative_mean, negative_variance, classifier_accuracy, df):
"""Simulate a noisy binary classifier by drawing per-cell scores from two Beta distributions.
Used inside the spacr screen simulator to model an imperfect
phenotype classifier: rows with ``is_active == 1`` should score high
and rows with ``is_active == 0`` should score low, but with
probability ``1 - classifier_accuracy`` a row is drawn from the
wrong Beta (simulating classifier error). The result is a
``score`` column on the input DataFrame ready to be fed to
:func:`compute_roc_auc` / :func:`compute_precision_recall`.
:param positive_mean: Mean of the Beta distribution for active
cells; must be in ``(0, 1)``.
:param positive_variance: Variance of the Beta for active cells;
must satisfy ``0 < var < mean * (1 - mean)``.
:param negative_mean: Mean of the Beta for inactive cells.
:param negative_variance: Variance of the Beta for inactive cells.
:param classifier_accuracy: Probability in ``[0, 1]`` that a row is
drawn from the Beta matching its true ``is_active`` label.
:param df: DataFrame containing an ``is_active`` column (0 / 1).
:returns: The input DataFrame with a new ``score`` column in
``[0, 1]``.
:raises ValueError: if either mean is outside ``(0, 1)`` or either
variance is outside ``(0, mean * (1 - mean))``.
Example:
.. code-block:: python
from spacr.sim import classifier
df = classifier(
positive_mean=0.8, positive_variance=0.02,
negative_mean=0.2, negative_variance=0.02,
classifier_accuracy=0.9, df=cells,
)
See Also:
:func:`compute_roc_auc`, :func:`compute_precision_recall`.
"""
def calc_alpha_beta(mean, variance):
"""Return ``(alpha, beta)`` for a Beta with the given mean and variance."""
if mean <= 0 or mean >= 1:
raise ValueError("Mean must be between 0 and 1 exclusively.")
max_variance = mean * (1 - mean)
if variance <= 0 or variance >= max_variance:
raise ValueError(f"Variance must be positive and less than {max_variance}.")
alpha = mean * (mean * (1 - mean) / variance - 1)
beta = alpha * (1 - mean) / mean
return alpha, beta
def get_score(is_active):
"""Return a Beta sample from the correct or incorrect class distribution."""
if np.random.rand() < classifier_accuracy:
return np.random.beta(a1, b1) if is_active else np.random.beta(a2, b2)
else:
return np.random.beta(a2, b2) if is_active else np.random.beta(a1, b1)
a1, b1 = calc_alpha_beta(positive_mean, positive_variance)
a2, b2 = calc_alpha_beta(negative_mean, negative_variance)
df['score'] = df['is_active'].apply(get_score)
return df
[docs]
def compute_roc_auc(cell_scores):
"""Return ROC-curve arrays and AUC for a DataFrame of ``is_active``/``score`` rows.
:param cell_scores: DataFrame with columns ``is_active`` and ``score``.
:returns: Dict with keys ``threshold``, ``tpr``, ``fpr``, ``roc_auc``.
"""
fpr, tpr, thresh = roc_curve(cell_scores['is_active'], cell_scores['score'], pos_label=1)
roc_auc = auc(fpr, tpr)
cell_roc_dict = {'threshold':thresh,'tpr': tpr,'fpr': fpr, 'roc_auc':roc_auc}
return cell_roc_dict
[docs]
def compute_precision_recall(cell_scores):
"""Return precision/recall/F1/PR-AUC arrays for a DataFrame of ``is_active``/``score`` rows.
:param cell_scores: DataFrame with columns ``is_active`` and ``score``.
:returns: Dict with keys ``threshold``, ``precision``, ``recall``,
``f1_score``, ``pr_auc``.
"""
pr, re, th = precision_recall_curve(cell_scores['is_active'], cell_scores['score'])
th = np.append(th, 1.0)
denominator = pr + re
f1_score = np.divide(
2 * pr * re,
denominator,
out=np.zeros_like(denominator, dtype=float),
where=denominator > 0,
)
pr_auc = auc(re, pr)
cell_pr_dict = {'threshold':th,'precision': pr,'recall': re, 'f1_score':f1_score, 'pr_auc': pr_auc}
return cell_pr_dict
[docs]
def get_optimum_threshold(cell_pr_dict):
"""Return the classification threshold that maximises F1 in a PR result dict.
:param cell_pr_dict: Dict as returned by :func:`compute_precision_recall`.
:returns: Threshold value that maximises the F1 score.
"""
cell_pr_dict_df = pd.DataFrame(cell_pr_dict)
max_x = cell_pr_dict_df.loc[cell_pr_dict_df['f1_score'].idxmax()]
optimum = float(max_x['threshold'])
return optimum
[docs]
def update_scores_and_get_cm(cell_scores, optimum):
"""Add a per-threshold predicted-label column and return the confusion matrix.
:param cell_scores: DataFrame with columns ``is_active`` and ``score``.
:param optimum: Score threshold used to binarise predictions.
:returns: Tuple ``(cell_scores, cell_cm)`` where ``cell_cm`` is a NumPy
confusion matrix.
"""
cell_scores[optimum] = cell_scores.score.map(lambda x: 1 if x >= optimum else 0)
cell_cm = metrics.confusion_matrix(cell_scores.is_active, cell_scores[optimum])
return cell_scores, cell_cm
[docs]
def cell_level_roc_auc(cell_scores):
"""Compute cell-level ROC/PR metrics and confusion matrix at the F1-optimal threshold.
:param cell_scores: DataFrame with columns ``is_active`` and ``score``.
:returns: Tuple ``(cell_roc_dict_df, cell_pr_dict_df, cell_scores, cell_cm)``.
"""
cell_roc_dict = compute_roc_auc(cell_scores)
cell_pr_dict = compute_precision_recall(cell_scores)
optimum = get_optimum_threshold(cell_pr_dict)
cell_scores, cell_cm = update_scores_and_get_cm(cell_scores, optimum)
cell_pr_dict['optimum'] = optimum
cell_roc_dict_df = pd.DataFrame(cell_roc_dict)
cell_pr_dict_df = pd.DataFrame(cell_pr_dict)
return cell_roc_dict_df, cell_pr_dict_df, cell_scores, cell_cm
[docs]
def generate_well_score(cell_scores):
"""Aggregate cell-level scores into per-well summary rows.
:param cell_scores: DataFrame indexed by cells with ``plate_row_column``,
``is_active`` and ``gene_id`` columns.
:returns: DataFrame indexed by ``plate_row_column`` with
``average_active_score``, ``gene_list``, and ``score`` columns.
"""
well_score = cell_scores.groupby(['plate_row_column']).agg(
average_active_score=('is_active', 'mean'),
gene_list=('gene_id', lambda x: np.unique(x).tolist()))
well_score['score'] = np.log10(well_score['average_active_score'] + 1)
return well_score
[docs]
def sequence_plates(well_score, number_of_genes, avg_reads_per_gene, sd_reads_per_gene, sequencing_error=0.01):
"""Simulate sequencing of every well and return per-well gene fraction and metadata.
Each gene present in a well accrues a Poisson-distributed read count that
may be reassigned to a random well with probability ``sequencing_error``.
:param well_score: DataFrame with a ``gene_list`` column per well.
:param number_of_genes: Number of distinct genes in the pool.
:param avg_reads_per_gene: Mean of the per-gene read count distribution.
:param sd_reads_per_gene: Standard deviation of that distribution.
:param sequencing_error: Probability of assigning a read to the wrong well.
Default ``0.01``.
:returns: Tuple ``(gene_fraction_map, metadata)`` DataFrames indexed by well.
"""
reads, _ = dist_gen(avg_reads_per_gene, sd_reads_per_gene, well_score)
gene_names = [f'gene_{v}' for v in range(number_of_genes+1)]
all_wells = well_score.index
gene_counts_map = pd.DataFrame(np.zeros((len(all_wells), number_of_genes+1)), columns=gene_names, index=all_wells)
for _, row in well_score.iterrows():
gene_list = row['gene_list']
if gene_list:
for gene in gene_list:
gene_count = int(random.choice(reads))
error = np.random.binomial(1, sequencing_error)
if error:
wrong_well = np.random.choice(all_wells)
gene_counts_map.loc[wrong_well, f'gene_{int(gene)}'] += gene_count
else:
gene_counts_map.loc[_, f'gene_{int(gene)}'] += gene_count
gene_fraction_map = gene_counts_map.div(gene_counts_map.sum(axis=1), axis=0)
gene_fraction_map = gene_fraction_map.fillna(0)
metadata = pd.DataFrame(index=well_score.index)
metadata['genes_in_well'] = gene_fraction_map.astype(bool).sum(axis=1)
metadata['sum_fractions'] = gene_fraction_map.sum(axis=1)
metadata['sum_reads'] = gene_counts_map.sum(axis=1)
return gene_fraction_map, metadata
[docs]
def regression_roc_auc(results_df, active_gene_list, control_gene_list, alpha = 0.05, optimal=False):
"""Score regression hits against ground truth and compute ROC/PR metrics.
Marks each gene as active/inactive/control, derives a hit cutoff from the
control coefficients, and returns ROC/PR curves, a confusion matrix and
per-run summary statistics.
:param results_df: Regression output with ``gene``, ``coef`` and ``P>|t|``.
:param active_gene_list: Gene indices considered truly active.
:param control_gene_list: Gene indices used to derive the coefficient cutoff.
:param alpha: Significance threshold applied to p-values. Default ``0.05``.
:param optimal: When True, use the F1-optimal probability threshold instead
of ``0.5`` for the final confusion matrix.
:returns: Tuple ``(results_df, reg_roc_dict_df, reg_pr_dict_df, reg_cm,
sim_stats)`` where ``sim_stats`` is a single-row DataFrame.
"""
results_df = results_df.rename(columns={"P>|t|": "p"})
actives_list = ['gene_' + str(i) for i in active_gene_list]
results_df['active'] = results_df['gene'].apply(lambda x: 1 if x in actives_list else 0)
results_df['active'] = results_df['active'].fillna(0)
controls_list = ['gene_' + str(i) for i in control_gene_list]
results_df['color'] = results_df['gene'].apply(lambda x: 'control' if x in controls_list else ('active' if x in actives_list else 'inactive'))
results_df['size'] = results_df['active']
results_df['p'] = results_df['p'].clip(lower=0.0001)
results_df['logp'] = -np.log10(results_df['p'])
control_df = results_df[results_df['color'] == 'control']
control_mean = control_df['coef'].mean()
control_var = control_df['coef'].var()
cutoff = abs(control_mean)+(3*control_var)
active_df = results_df[results_df['color'] == 'active']
active_mean = active_df['coef'].mean()
active_std = active_df['coef'].std()
active_var = active_df['coef'].var()
inactive_df = results_df[results_df['color'] == 'inactive']
inactive_mean = inactive_df['coef'].mean()
inactive_std = inactive_df['coef'].std()
inactive_var = inactive_df['coef'].var()
results_df['score'] = np.where(((results_df['coef'] >= cutoff) | (results_df['coef'] <= -cutoff)) & (results_df['p'] <= alpha), 1, 0)
fpr, tpr, thresh = roc_curve(results_df['active'], results_df['score'])
roc_auc = auc(fpr, tpr)
reg_roc_dict_df = pd.DataFrame({'threshold':thresh, 'tpr': tpr, 'fpr': fpr, 'roc_auc':roc_auc})
pr, re, th = precision_recall_curve(results_df['active'], results_df['score'])
th = np.append(th, 1.0)
denominator = pr + re
f1_score = np.divide(
2 * pr * re,
denominator,
out=np.zeros_like(denominator, dtype=float),
where=denominator > 0,
)
pr_auc = auc(re, pr)
reg_pr_dict_df = pd.DataFrame({'threshold':th, 'precision': pr, 'recall': re, 'f1_score':f1_score, 'pr_auc': pr_auc})
optimal_row = reg_pr_dict_df['f1_score'].idxmax()
optimal_threshold = float(reg_pr_dict_df.loc[optimal_row, 'threshold'])
if optimal:
results_df[optimal_threshold] = results_df.score.apply(lambda x: 1 if x >= optimal_threshold else 0)
reg_cm = confusion_matrix(results_df.active, results_df[optimal_threshold])
else:
results_df[0.5] = results_df.score.apply(lambda x: 1 if x >= 0.5 else 0)
reg_cm = confusion_matrix(results_df.active, results_df[0.5])
TN = reg_cm[0][0]
FP = reg_cm[0][1]
FN = reg_cm[1][0]
TP = reg_cm[1][1]
accuracy = (TP + TN) / (TP + FP + FN + TN)
sim_stats = {'optimal_threshold':optimal_threshold,
'accuracy': accuracy,
'prauc':pr_auc,
'roc_auc':roc_auc,
'inactive_mean':inactive_mean,
'inactive_std':inactive_std,
'inactive_var':inactive_var,
'active_mean':active_mean,
'active_std':active_std,
'active_var':active_var,
'cutoff':cutoff,
'TP':TP,
'FP':FP,
'TN':TN,
'FN':FN}
return results_df, reg_roc_dict_df, reg_pr_dict_df, reg_cm, pd.DataFrame([sim_stats])
[docs]
def plot_histogram(data, x_label, ax, color, title, binwidth=0.01, log=False):
"""Draw a Seaborn density histogram on ``ax`` for the given column.
:param data: Data source passed to ``sns.histplot``.
:param x_label: Column name plotted on the x-axis.
:param ax: Matplotlib axes to draw into.
:param color: Bar/fill color.
:param title: Axes title.
:param binwidth: Histogram bin width; falsy uses Seaborn's default.
:param log: When True, apply a log scale to the y-axis.
:returns: None.
"""
if not binwidth:
sns.histplot(data=data, x=x_label, ax=ax, color=color, kde=False, stat='density',
legend=False, fill=True, element='step')
else:
sns.histplot(data=data, x=x_label, ax=ax, color=color, binwidth=binwidth, kde=False, stat='density',
legend=False, fill=True, element='step')
if log:
ax.set_yscale('log')
ax.set_title(title)
ax.set_xlabel(x_label)
[docs]
def plot_roc_pr(data, ax, title, x_label, y_label):
"""Plot a ROC or PR curve with a diagonal random-classifier reference line.
:param data: DataFrame containing the ``x_label`` and ``y_label`` columns.
:param ax: Matplotlib axes to draw into.
:param title: Axes title.
:param x_label: Column name plotted on the x-axis.
:param y_label: Column name plotted on the y-axis.
:returns: None.
"""
ax.plot(data[x_label], data[y_label], color=SIM_ACTIVE, lw=1.2)
ax.plot([0, 1], [0, 1], color=ROLES['reference'], lw=0.6,
linestyle=(0, (4, 3)), label='random classifier', zorder=0)
ax.set_title(title)
ax.set_ylabel(y_label)
ax.set_xlabel(x_label)
ax.legend(loc="lower right", frameon=False)
[docs]
def plot_confusion_matrix(data, ax, title):
"""Render a 2x2 confusion matrix as an annotated Seaborn heatmap.
:param data: 2x2 NumPy confusion matrix ordered ``[[TN, FP], [FN, TP]]``.
:param ax: Matplotlib axes to draw into.
:param title: Axes title.
:returns: None.
"""
group_names = ['True Neg','False Pos','False Neg','True Pos']
group_counts = ["{0:0.0f}".format(value) for value in data.flatten()]
group_percentages = ["{0:.2%}".format(value) for value in data.flatten()/np.sum(data)]
sns.heatmap(data, cmap=Palette.SEQUENTIAL, ax=ax)
ink = resolve_ink(theme_target())
for i in range(data.shape[0]):
for j in range(data.shape[1]):
ax.text(j+0.5, i+0.5, f'{group_names[i*2+j]}\n{group_counts[i*2+j]}\n{group_percentages[i*2+j]}',
ha="center", va="center", color=ink,
fontsize=TYPE_SCALE['annotation'])
ax.set_title(title)
ax.set_xlabel('\nPredicted Values')
ax.set_ylabel('Actual Values ')
ax.xaxis.set_ticklabels(['False','True'])
ax.yaxis.set_ticklabels(['False','True'])
[docs]
def run_simulation(settings):
"""Run one end-to-end pooled-screen simulation and return every intermediate table.
Composes :func:`generate_gene_list`, :func:`generate_plate_map`,
:func:`run_experiment`, :func:`classifier`, cell/well aggregation,
:func:`sequence_plates` and :func:`regression_roc_auc` into a single call.
:param settings: Dict of simulation parameters (gene counts, distribution
moments, sequencing error, classifier accuracy, ...).
:returns: Tuple ``(cell_scores, cell_roc_dict_df, cell_pr_dict_df,
cell_cm, well_score, gene_fraction_map, metadata, results_df,
reg_roc_dict_df, reg_pr_dict_df, reg_cm, sim_stats,
genes_per_well_df, wells_per_gene_df, dists)``.
"""
active_gene_list = generate_gene_list(settings['number_of_active_genes'], settings['number_of_genes'])
control_gene_list = generate_gene_list(settings['number_of_control_genes'], settings['number_of_genes'])
plate_map = generate_plate_map(settings['nr_plates'])
control_columns = ['1', '2', '3', '23', '24']
plate_map = plate_map[~plate_map['column_id'].isin(control_columns)].reset_index(drop=True)
cell_level, genes_per_well_df, wells_per_gene_df, dists = run_experiment(plate_map, settings['number_of_genes'], active_gene_list, settings['avg_genes_per_well'], settings['sd_genes_per_well'], settings['avg_cells_per_well'], settings['sd_cells_per_well'], settings['well_ineq_coeff'], settings['gene_ineq_coeff'])
cell_scores = classifier(settings['positive_mean'], settings['positive_variance'], settings['negative_mean'], settings['negative_variance'], settings['classifier_accuracy'], df=cell_level)
cell_roc_dict_df, cell_pr_dict_df, cell_scores, cell_cm = cell_level_roc_auc(cell_scores)
well_score = generate_well_score(cell_scores)
gene_fraction_map, metadata = sequence_plates(well_score, settings['number_of_genes'], settings['avg_reads_per_gene'], settings['sd_reads_per_gene'], sequencing_error=settings['sequencing_error'])
x = gene_fraction_map
y = np.log10(well_score['score']+1)
x = sm.add_constant(x)
model = sm.OLS(y, x).fit()
results_summary = model.summary()
results_as_html = results_summary.tables[1].as_html()
results_df = pd.read_html(StringIO(results_as_html), header=0, index_col=0)[0]
results_df = results_df.rename_axis("gene").reset_index()
results_df = results_df.iloc[1: , :]
results_df, reg_roc_dict_df, reg_pr_dict_df, reg_cm, sim_stats = regression_roc_auc(results_df, active_gene_list, control_gene_list, alpha = 0.05, optimal=False)
output = [cell_scores, cell_roc_dict_df, cell_pr_dict_df, cell_cm, well_score, gene_fraction_map, metadata, results_df, reg_roc_dict_df, reg_pr_dict_df, reg_cm, sim_stats, genes_per_well_df, wells_per_gene_df]
del cell_scores, cell_roc_dict_df, cell_pr_dict_df, cell_cm, well_score, gene_fraction_map, metadata, results_df, reg_roc_dict_df, reg_pr_dict_df, reg_cm, sim_stats, genes_per_well_df, wells_per_gene_df
gc.collect()
return output, dists
[docs]
def vis_dists(dists, src, v, i):
"""Save side-by-side histograms of the six per-run distributions in ``dists``.
:param dists: Six arrays in order ``[genes/well, wells/gene, gini_well,
gini_gene, gene_weights, well_weights]``.
:param src: Output directory used by :func:`save_plot`.
:param v: Variable label passed through to :func:`save_plot`.
:param i: Simulation index passed through to :func:`save_plot`.
:returns: None.
"""
n_graphs = 6
height_graphs = 4
n=0
width_graphs = height_graphs*n_graphs
with figure_style(theme_target()):
fig2, ax =plt.subplots(1,n_graphs, figsize = (width_graphs,height_graphs))
names = ['genes/well', 'wells/gene', 'genes/well gini', 'wells/gene gini', 'gene_weights', 'well_weights']
from .figures.bundle import _register_figure_data
_register_figure_data(fig2, dict(zip(names, dists)), x="distribution", y="value", kind="hist")
for index, dist in enumerate(dists):
temp = pd.DataFrame(dist, columns = [f'{names[index]}'])
sns.histplot(data=temp, x=f'{names[index]}', kde=False, binwidth=None, stat='count', element="step", ax=ax[n], color=ROLES['fill'], log_scale=False)
n+=1
save_plot(fig2, src, 'dists', i)
plt.close(fig2)
with figure_style(theme_target()):
plt.cla()
plt.clf()
del dists
return
[docs]
def visualize_all(output):
"""Render the full 13-panel diagnostic figure for one simulation output.
:param output: The 14-element list returned by :func:`run_simulation` (all
elements before ``dists``).
:returns: The generated Matplotlib figure.
"""
cell_scores = output[0]
cell_roc_dict_df = output[1]
cell_pr_dict_df = output[2]
cell_cm = output[3]
well_score = output[4]
results_df = output[7]
reg_roc_dict_df = output[8]
reg_pr_dict_df = output[9]
reg_cm =output[10]
genes_per_well_df = output[12]
wells_per_gene_df = output[13]
hline = -np.log10(0.05)
n_graphs = 13
height_graphs = 4
n=0
width_graphs = height_graphs*n_graphs
with figure_style(theme_target()):
fig, ax =plt.subplots(1,n_graphs, figsize = (width_graphs,height_graphs))
from .figures.bundle import _register_figure_data
_register_figure_data(fig, genes_per_well_df, y="genes_per_well", kind="hist")
gini_genes_per_well = gini(genes_per_well_df['genes_per_well'].tolist())
plot_histogram(genes_per_well_df, "genes_per_well", ax[n], SIM_INACTIVE, f'gene/well (gini = {gini_genes_per_well:.2f})', binwidth=None, log=False)
n+=1
gini_wells_per_gene = gini(wells_per_gene_df['wells_per_gene'].tolist())
plot_histogram(wells_per_gene_df, "wells_per_gene", ax[n], SIM_INACTIVE, f'well/gene (Gini = {gini_wells_per_gene:.2f})', binwidth=None, log=False)
n+=1
active_distribution = cell_scores[cell_scores['is_active'] == 1]
inactive_distribution = cell_scores[cell_scores['is_active'] == 0]
plot_histogram(active_distribution, "score", ax[n], SIM_ACTIVE, 'Cell scores', log=False)
plot_histogram(inactive_distribution, "score", ax[n], SIM_INACTIVE, 'Cell scores', log=False)
legend_elements = [Patch(facecolor=SIM_ACTIVE, edgecolor=SIM_ACTIVE, label='Active'),
Patch(facecolor=SIM_INACTIVE, edgecolor=SIM_INACTIVE, label='Inactive')]
ax[n].legend(handles=legend_elements, loc='upper right', frameon=False)
ax[n].set_xlim([0, 1])
n+=1
inactive_distribution_well = inactive_distribution.groupby(['plate_id', 'row_id', 'column_id'])['score'].mean().reset_index(name='score')
active_distribution_well = active_distribution.groupby(['plate_id', 'row_id', 'column_id'])['score'].mean().reset_index(name='score')
mixed_distribution_well = cell_scores.groupby(['plate_id', 'row_id', 'column_id'])['score'].mean().reset_index(name='score')
plot_histogram(inactive_distribution_well, "score", ax[n], SIM_INACTIVE, 'Well scores', log=False)
plot_histogram(active_distribution_well, "score", ax[n], SIM_ACTIVE, 'Well scores', log=False)
plot_histogram(mixed_distribution_well, "score", ax[n], SIM_MIXED, 'Well scores', log=False)
legend_elements = [Patch(facecolor=SIM_INACTIVE, edgecolor=SIM_INACTIVE, label='Inactive'),
Patch(facecolor=SIM_ACTIVE, edgecolor=SIM_ACTIVE, label='Active'),
Patch(facecolor=SIM_MIXED, edgecolor=SIM_MIXED, label='Mixed')]
ax[n].legend(handles=legend_elements, loc='upper right', frameon=False)
ax[n].set_xlim([0, 1])
n+=1
plot_roc_pr(cell_roc_dict_df, ax[n], 'ROC (Cell)', 'fpr', 'tpr')
n+=1
plot_roc_pr(cell_pr_dict_df, ax[n], 'Precision recall (Cell)', 'recall', 'precision')
ax[n].set_ylim([-0.1, 1.1])
ax[n].set_xlim([-0.1, 1.1])
n+=1
plot_confusion_matrix(cell_cm, ax[n], 'Confusion Matrix Cell')
n+=1
plot_histogram(well_score, "score", ax[n], ROLES['fill'], 'Well score', binwidth=0.005, log=True)
n+=1
control_df = results_df[results_df['color'] == 'control']
control_mean = control_df['coef'].mean()
control_var = control_df['coef'].std()
cutoff = abs(control_mean)+(3*control_var)
categories = ['inactive', 'control', 'active']
colors = [SIM_INACTIVE, SIM_CONTROL, SIM_ACTIVE]
for category, color in zip(categories, colors):
df = results_df[results_df['color'] == category]
ax[n].scatter(df['coef'], df['logp'], c=color, alpha=0.7, label=category,
linewidths=0)
reg_lab = ax[n].legend(title='', frameon=False,
prop={'size': TYPE_SCALE['legend']})
ax[n].add_artist(reg_lab)
reference_line(ax[n], y=hline)
reference_line(ax[n], x=-cutoff)
reference_line(ax[n], x=cutoff)
ax[n].set_title(f'Regression, threshold {cutoff:.3f}')
ax[n].set_xlim([-1, 1.1])
n+=1
df = results_df[['gene', 'coef', 'std err', 'p']].copy()
df = df.sort_values(by = ['coef', 'p'], ascending = [True, False], na_position = 'first')
df['rank'] = [*range(0,len(df),1)]
ax[n].plot(df['rank'], df['coef'], '-', color=Palette.GREY_DARK, lw=1.2)
ax[n].fill_between(df['rank'], df['coef'] - abs(df['std err']), df['coef'] + abs(df['std err']), alpha=0.25, color=Palette.GREY_DARK, linewidth=0)
ax[n].set_title('Effect score error')
ax[n].set_xlabel('rank')
ax[n].set_ylabel('Effect size')
n+=1
plot_roc_pr(reg_roc_dict_df, ax[n], 'ROC (gene)', 'fpr', 'tpr')
ax[n].legend(loc="lower right", frameon=False)
n+=1
plot_roc_pr(reg_pr_dict_df, ax[n], 'Precision recall (gene)', 'recall', 'precision')
ax[n].legend(loc="lower right", frameon=False)
n+=1
plot_confusion_matrix(reg_cm, ax[n], 'Confusion Matrix Reg')
for n in [*range(0,n_graphs,1)]:
ax[n].spines['top'].set_visible(False)
ax[n].spines['right'].set_visible(False)
plt.tight_layout()
plt.show()
gc.collect()
return fig
[docs]
def create_database(db_path):
"""Ensure a SQLite database file exists at ``db_path``.
:param db_path: Filesystem path for the SQLite database.
:returns: None.
"""
conn = None
try:
from .database_concurrency import connect as _connect_database
conn = _connect_database(db_path)
except Exception as e:
print(e)
finally:
if conn:
conn.close()
[docs]
def append_database(src, table, table_name):
"""Append a DataFrame to ``<src>/simulations.db`` under ``table_name``.
:param src: Directory containing (or that should contain) ``simulations.db``.
:param table: DataFrame written with ``if_exists='append'``.
:param table_name: Target table name in the SQLite database.
:returns: None.
"""
from .database_concurrency import _capture_write_operation, _inside_write_packet
if _capture_write_operation('simulation', (src, table, table_name)):
return
path = os.path.join(src, 'simulations.db')
try:
from .tabular import write_database
if not os.path.isdir(src):
raise sqlite3.OperationalError('unable to open database file')
write_database(table, path, table_name, if_exists='append',
canonicalise=False, index=False)
except sqlite3.OperationalError as e:
if _inside_write_packet(path):
raise
print("SQLite error:", e)
return
[docs]
def save_data(src, output, settings, save_all=False, i=0, variable='all'):
"""Persist one simulation's output tables to a SQLite database under ``src``.
In the default mode only a concatenated summary row (settings + sim_stats
+ Gini metrics) is appended to a ``simulations`` table. When ``save_all``
is True, every intermediate table is written under its canonical name.
:param src: Output directory containing ``simulations.db``.
:param output: 14-element list from :func:`run_simulation`.
:param settings: Simulation settings dict recorded as the first row.
:param save_all: When True, write every intermediate table separately.
:param i: Simulation index used to tag the summary row.
:param variable: Name of the swept variable used for tagging.
:returns: None.
"""
try:
if not save_all:
src = f'{src}'
os.makedirs(src, exist_ok=True)
else:
os.makedirs(src, exist_ok=True)
settings_df = pd.DataFrame({key: [value] for key, value in settings.items()})
output = [settings_df] + output
table_names = ['settings', 'cell_scores', 'cell_roc', 'cell_precision_recall', 'cell_confusion_matrix', 'well_score', 'gene_fraction_map', 'metadata', 'regression_results', 'regression_roc', 'regression_precision_recall', 'regression_confusion_matrix', 'sim_stats', 'genes_per_well', 'wells_per_gene']
if not save_all:
gini_genes_per_well = gini(output[13]['genes_per_well'].tolist())
gini_wells_per_gene = gini(output[14]['wells_per_gene'].tolist())
indices_to_keep= [0,12]
filtered_output = [v for i, v in enumerate(output) if i in indices_to_keep]
df_concat = pd.concat(filtered_output, axis=1)
df_concat['genes_per_well_gini'] = gini_genes_per_well
df_concat['wells_per_gene_gini'] = gini_wells_per_gene
df_concat['date'] = datetime.now()
df_concat[f'variable_{variable}_sim_nr'] = i
append_database(src, df_concat, 'simulations')
del gini_genes_per_well, gini_wells_per_gene, df_concat
if save_all:
for i, df in enumerate(output):
df = output[i]
if table_names[i] == 'well_score':
df['gene_list'] = df['gene_list'].astype(str)
if not isinstance(df, pd.DataFrame):
df = pd.DataFrame(df)
append_database(src, df, table_names[i])
del df
except Exception as e:
from .database_concurrency import _WRITE_PACKET_OPERATIONS
if _WRITE_PACKET_OPERATIONS.get() is not None:
raise
print(f"An error occurred while saving data: {e}")
print(traceback.format_exc())
return
[docs]
def save_plot(fig, src, variable, i):
"""Save a Matplotlib figure to ``<src>/<variable>/<i>_figure``.
The output format and resolution follow the user's figure preferences.
A default installation writes PDF.
:param fig: Figure to save.
:param src: Root directory for outputs.
:param variable: Sub-folder name (the swept variable label).
:param i: Zero-padded simulation index used in the file name.
:returns: None.
"""
os.makedirs(f'{src}/{variable}', exist_ok=True)
filename_fig = f'{src}/{variable}/{str(i)}_figure.pdf'
save_figure(fig, filename_fig, bbox_inches='tight')
return
[docs]
def run_and_save(i, settings, time_ls, total_sims):
"""Worker that runs one simulation, saves outputs, and appends its runtime.
:param i: Simulation index (used for filenames and tagging).
:param settings: Simulation settings dict.
:param time_ls: Shared list receiving the elapsed time in seconds.
:param total_sims: Total simulation count (used for progress display only).
:returns: Tuple ``(i, sim_time, None)``.
"""
if settings.get('random_seed'):
random.seed(42)
src = settings['src']
plot = settings['plot']
v = settings['variable']
start_time = time()
date_string = settings['start_time']
output, dists = run_simulation(settings)
sim_time = time() - start_time
settings['sim_time'] = sim_time
src = os.path.join(f'{src}/{date_string}',settings['name'])
save_data(src, output, settings, save_all=False, i=i, variable=v)
if plot:
vis_dists(dists,src, v, i)
fig = visualize_all(output)
save_plot(fig, src, v, i)
plt.close(fig)
with figure_style(theme_target()):
plt.cla()
plt.clf()
del fig
del output, dists
gc.collect()
time_ls.append(sim_time)
return i, sim_time, None
_SIMULATION_WRITE_ENDPOINT = None
def _initialize_simulation_writer(endpoint):
"""Give simulation workers bounded producer handles for one SQL writer."""
global _SIMULATION_WRITE_ENDPOINT
_SIMULATION_WRITE_ENDPOINT = endpoint
def _run_and_save_queued(i, settings, time_ls, total_sims):
"""Capture one complete simulation before enqueuing its atomic save."""
import json
from .database_concurrency import _capture_write_packet
if _SIMULATION_WRITE_ENDPOINT is None:
raise RuntimeError('Simulation worker has no database writer endpoint')
with _capture_write_packet() as operations:
result = run_and_save(i, dict(settings), [], total_sims)
path = os.path.abspath(os.path.join(settings['src'], settings['start_time'],
settings['name'], 'simulations.db'))
_SIMULATION_WRITE_ENDPOINT.enqueue(json.dumps([path, i]), operations)
time_ls.append(result[1])
return result
def _commit_simulation_packet(packet):
"""Save every table and the simulation ticket in the same transaction."""
from .database_concurrency import _commit_write_packet
import json
path, index = json.loads(packet['field'])
if not isinstance(path, str) or not isinstance(index, int):
raise ValueError('Invalid simulation packet identity')
paths = set()
for name, arguments, keywords in packet['operations'] or ():
if name != 'simulation' or keywords:
raise ValueError('Unsupported simulation database operation')
paths.add(os.path.abspath(os.path.join(arguments[0], 'simulations.db')))
if paths != {path} and packet['operations'] is not None:
raise ValueError('A simulation packet must target exactly one database')
def dispatch(operation):
"""Write one captured simulation frame through the normal table writer."""
append_database(*operation[1])
return _commit_write_packet(path, packet, dispatch)
[docs]
def validate_and_adjust_beta_params(sim_params):
"""Clamp per-run Beta variances so the requested mean/variance is feasible.
:param sim_params: List of per-run parameter dicts with ``positive_mean``,
``negative_mean``, ``positive_variance``, ``negative_variance``.
:returns: The same list with any infeasible variances capped to 99% of the
theoretical maximum for the requested mean.
"""
adjusted_params = []
for params in sim_params:
max_pos_variance = params['positive_mean'] * (1 - params['positive_mean'])
max_neg_variance = params['negative_mean'] * (1 - params['negative_mean'])
if params['positive_variance'] >= max_pos_variance:
print(f'changed positive variance from {params["positive_variance"]} to {max_pos_variance * 0.99}')
params['positive_variance'] = max_pos_variance * 0.99
if params['negative_variance'] >= max_neg_variance:
print(f'changed negative variance from {params["negative_variance"]} to {max_neg_variance * 0.99}')
params['negative_variance'] = max_neg_variance * 0.99
adjusted_params.append(params)
return adjusted_params
def _simulation_unit_bytes(settings):
"""Approximate bytes one simulation holds: a per-cell table for its plates.
The largest swept plate count, cells per well and genes per well give
rows of eight-byte values for 384 wells per plate.
:returns: ``0`` when the sweep values cannot be read.
"""
def _largest(key, default):
"""The largest value swept for ``key``."""
value = settings.get(key, default)
values = value if isinstance(value, (list, tuple)) else [value]
try:
return max(float(v) for v in values)
except (TypeError, ValueError):
return float(default)
try:
rows = (_largest('nr_plates', 1) * 384
* _largest('avg_cells_per_well', 100))
return int(rows * (_largest('avg_genes_per_well', 2) + 8) * 8)
except Exception:
return 0
[docs]
def generate_parameters(settings):
"""Expand a sweep-settings dict into one settings dict per (Cartesian) simulation.
:param settings: Config dict where each swept key holds an iterable of values.
:returns: Shuffled list of per-run settings dicts, already run through
:func:`validate_and_adjust_beta_params`.
"""
if not isinstance(settings.get('positive_mean'), (list, tuple)):
settings['positive_mean'] = [0.8]
sim_ls = []
for avg_genes_per_well in settings['avg_genes_per_well']:
replicates = settings['replicates']
for avg_cells_per_well in settings['avg_cells_per_well']:
for classifier_accuracy in settings['classifier_accuracy']:
for positive_mean in settings['positive_mean']:
for avg_reads_per_gene in settings['avg_reads_per_gene']:
for sequencing_error in settings['sequencing_error']:
for well_ineq_coeff in settings['well_ineq_coeff']:
for gene_ineq_coeff in settings['gene_ineq_coeff']:
for nr_plates in settings['nr_plates']:
for number_of_genes in settings['number_of_genes']:
for number_of_active_genes in settings['number_of_active_genes']:
for i in range(1, replicates+1):
sett = deepcopy(settings)
sett['avg_genes_per_well'] = avg_genes_per_well
sett['sd_genes_per_well'] = avg_genes_per_well / 2
sett['avg_cells_per_well'] = avg_cells_per_well
sett['sd_cells_per_well'] = avg_cells_per_well / 2
sett['classifier_accuracy'] = classifier_accuracy
sett['positive_mean'] = positive_mean
sett['negative_mean'] = 1-positive_mean
sett['positive_variance'] = (1-positive_mean)/2
sett['negative_variance'] = (1-positive_mean)/2
sett['avg_reads_per_gene'] = avg_reads_per_gene
sett['sd_reads_per_gene'] = avg_reads_per_gene / 2
sett['sequencing_error'] = sequencing_error
sett['well_ineq_coeff'] = well_ineq_coeff
sett['gene_ineq_coeff'] = gene_ineq_coeff
sett['nr_plates'] = nr_plates
sett['number_of_genes'] = number_of_genes
sett['number_of_active_genes'] = number_of_active_genes
sim_ls.append(sett)
random.shuffle(sim_ls)
sim_ls = validate_and_adjust_beta_params(sim_ls)
print(f'Running {len(sim_ls)} simulations.')
return sim_ls
#: Backward-compatible alias for the historical misspelling of
#: :func:`generate_parameters`.
generate_paramiters = generate_parameters
[docs]
def run_multiple_simulations(settings):
"""Fan out the sweep from :func:`generate_parameters` across a process pool.
Processing workers start ten seconds apart and drive :func:`run_and_save`
with captured database operations. One parent-owned writer commits each
complete simulation and its retry ticket atomically. Explicit resource
overloads receive one serial retry after all primary tasks finish; database
overloads use the writer's separate final queue. Invalid data and
cancellation are not retried. An unsuccessful simulation or final database
write raises instead of reporting a successful sweep.
:param settings: Sweep-settings dict. Must include ``max_workers``.
``database_write_queue_gib`` optionally overrides the saved queue RAM
allowance, from zero to 64 GiB; zero buffers serialized packets on disk.
Pending packets remain below the dated output folder on interruption.
:returns: None.
:raises ValueError: if the queue RAM allowance is invalid.
:raises RuntimeError: if a final database write fails. Processing errors
propagate with their original exception type.
"""
now = datetime.now()
start_time = now.strftime("%y%m%d")
settings['start_time'] = start_time
sim_ls = generate_parameters(settings)
max_workers = settings['max_workers'] or max(1, cpu_count() - 4)
from .resource_log import _guard_workers
max_workers = _guard_workers('simulation', max_workers,
_simulation_unit_bytes(settings),
settings=settings)
with Manager() as manager:
time_ls = manager.list()
total_sims = len(sim_ls)
from .database_concurrency import _DatabaseWriteQueue
budget = settings.get('database_write_queue_gib')
if budget is None:
from .qt.preferences import get_database_write_queue_gib
budget = get_database_write_queue_gib()
budget = float(budget)
if not 0 <= budget <= 64:
raise ValueError('Database write queue RAM must be between 0 and 64 GiB')
failures = []
writer = _DatabaseWriteQueue(
os.path.join(settings['src'], start_time, '.simulation_write_queue'),
_commit_simulation_packet,
lambda field, ticket, error: failures.append(error) if error else None,
ram_gib=budget)
try:
with Pool(max_workers, initializer=_initialize_simulation_writer,
initargs=(writer.endpoint,)) as pool:
writer.start()
result = pool.starmap_async(_run_and_save_queued, [
(index, simulation_settings, time_ls, total_sims)
for index, simulation_settings in enumerate(sim_ls)])
while not result.ready():
try:
sleep(0.01)
sims_processed = len(time_ls)
average_time = np.mean(time_ls) if len(time_ls) > 0 else 0
time_left = (((total_sims - sims_processed) * average_time) / max_workers) / 60
print(f'Progress: {sims_processed}/{total_sims} Time/simulation {average_time:.3f}sec Time Remaining {time_left:.3f} min.', end='\r', flush=True)
gc.collect()
except Exception as error:
print(error)
print(traceback.format_exc())
result.get()
pool.close()
pool.join()
writer.finish()
if failures:
raise RuntimeError('; '.join(str(error) for error in failures))
except BaseException:
if writer._thread is not None:
writer.cancel()
try:
writer.finish()
except BaseException:
pass
raise
[docs]
def generate_integers(start, stop, step):
"""Return ``list(range(start, stop + 1, step))`` (inclusive upper bound).
:param start: first integer in the generated sequence.
:param stop: inclusive upper endpoint for the sequence.
:param step: integer spacing passed to :class:`range`.
"""
return list(range(start, stop + 1, step))
[docs]
def generate_floats(start, stop, step):
"""Return floats from ``start`` through ``stop`` with ``step`` spacing.
:param start: first numeric value in the generated sequence.
:param stop: inclusive endpoint when reached by the requested spacing.
:param step: numeric spacing and source of output decimal precision.
"""
num_decimals = str(step)[::-1].find('.')
if num_decimals < 0:
num_decimals = 0
n_steps = int(math.floor(round((stop - start) / step, 9))) + 1
floats_list = [round(start + i * step, num_decimals) for i in range(max(n_steps, 0))]
return floats_list
[docs]
def remove_columns_with_single_value(df):
"""Return ``df`` without columns whose values are constant across rows.
:param df: Source DataFrame.
:returns: Copy of ``df`` with zero-variance columns dropped.
"""
to_drop = [column for column in df.columns if df[column].nunique() == 1]
return df.drop(to_drop, axis=1)
[docs]
def read_simulations_table(db_path):
"""Return the ``simulations`` table from ``db_path`` as a DataFrame.
:param db_path: Path to a SQLite database written by :func:`save_data`.
:returns: DataFrame of the ``simulations`` table, or ``None`` on failure.
"""
from .database_concurrency import connect as _connect_database
conn = _connect_database(db_path)
try:
df = pd.read_sql_query("SELECT * FROM simulations", conn)
except Exception as e:
print(f"An error occurred: {e}")
return None
finally:
conn.close()
return df
[docs]
def plot_simulations(df, variable, x_rotation=None, legend=False, grid=False, clean=True, verbose=False):
"""Grid-plot PR-AUC vs ``variable`` for every unique combination of the other sweep dimensions.
:param df: DataFrame containing ``prauc``, ``variable`` and the standard
grouping columns (``number_of_active_genes``, ``avg_reads_per_gene``, ...).
:param variable: Column plotted on the x-axis of each subplot.
:param x_rotation: Degrees to rotate x-tick labels. ``None`` uses 45.
:param legend: When True, show the per-subplot legend.
:param grid: When True, draw grid lines.
:param clean: When True, drop grouping columns whose values never vary.
:param verbose: When True, annotate each subplot with its filter conditions.
:returns: The generated Matplotlib figure.
"""
grouping_vars = ['number_of_active_genes', 'number_of_control_genes', 'avg_reads_per_gene',
'classifier_accuracy', 'nr_plates', 'number_of_genes', 'avg_genes_per_well',
'avg_cells_per_well', 'sequencing_error', 'well_ineq_coeff', 'gene_ineq_coeff']
grouping_vars = [col for col in grouping_vars if col != variable]
required_columns = {variable, 'prauc'} | set(grouping_vars)
if not required_columns.issubset(df.columns):
missing_cols = required_columns - set(df.columns)
raise ValueError(f"DataFrame must contain {missing_cols} columns")
if clean:
grouping_vars = [col for col in grouping_vars if df[col].nunique() > 1]
if grouping_vars:
unique_combinations = df[grouping_vars].drop_duplicates()
else:
unique_combinations = df.iloc[[0]][[]]
num_combinations = len(unique_combinations)
num_rows = math.ceil(np.sqrt(num_combinations))
num_cols = math.ceil(num_combinations / num_rows)
with figure_style(theme_target()):
fig, axes = plt.subplots(num_rows, num_cols, figsize=(5 * num_cols, 5 * num_rows))
from .figures.bundle import _register_figure_data
_register_figure_data(fig, df, x=variable, y="prauc" if "prauc" in df.columns else "", kind="line")
if num_rows * num_cols > 1:
axes = axes.flatten()
else:
axes = [axes]
idx = -1
for idx, (ax, (_, row)) in enumerate(zip(axes, unique_combinations.iterrows())):
condition = {var: row[var] for var in grouping_vars}
subset_df = df[df[grouping_vars].eq(row).all(axis=1)]
grouped = subset_df.groupby(variable)['prauc'].agg(['mean', 'std'])
grouped = grouped.sort_index()
ax.plot(grouped.index, grouped['mean'], marker='o', linestyle='-', color=SIM_ACTIVE, label='Mean PRAUC')
ax.fill_between(grouped.index, grouped['mean'] - grouped['std'], grouped['mean'] + grouped['std'], color=SIM_ACTIVE, alpha=0.25, linewidth=0, label='Std Dev')
ax.set_xlabel(variable)
ax.set_ylabel('Precision-Recall AUC (PRAUC)')
ax.grid(grid)
if legend:
ax.legend(frameon=False)
ax.set_xticks(grouped.index)
ax.set_xticklabels(grouped.index, rotation=x_rotation if x_rotation is not None else 45)
if verbose:
verbose_text = '\n'.join([f"{var}: {val}" for var, val in condition.items()])
ax.text(0.95, 0.05, verbose_text, transform=ax.transAxes,
fontsize=TYPE_SCALE['annotation'],
color=resolve_ink(theme_target()),
verticalalignment='bottom', horizontalalignment='right')
for ax in axes[idx+1:]:
ax.set_visible(False)
plt.tight_layout()
plt.show()
return fig
def _figures_dst(dst=None):
"""Resolve where a simulation figure is written.
The literal folder name was hard-coded at five call sites. Being a
RELATIVE path, every simulator run dropped a ``figures/`` tree into whatever
directory the process happened to be launched from -- the user's home,
a cluster scratch dir, or the repo itself, which is how a test run
leaked ``figures/feature_importance`` into the working tree.
:param dst: explicit destination; ``None`` falls back to the old
relative name so existing callers keep working, and the
``SPACR_SIM_FIGURES`` environment variable overrides both.
:returns: the directory to hand to ``save_plot``.
"""
if dst:
return str(dst)
return os.environ.get('SPACR_SIM_FIGURES', 'figures')
[docs]
def plot_correlation_matrix(df, annot=False, cmap=None, clean=True, dst=None):
"""Render a lower-triangular correlation heatmap of the standard sweep + metric columns.
:param df: DataFrame containing sweep variables plus ``prauc``, ``roc_auc``
and related outputs.
:param annot: When True, write numeric correlations in each cell.
:param cmap: Colormap name or object. ``None`` uses the house diverging
map, which is the one case the style allows a diverging map for --
a correlation is genuinely signed. The parameter used to default to
``'inferno'`` and then be OVERWRITTEN two lines later, so a caller
who passed one got the diverging map anyway and never knew.
:param clean: When True, drop constant columns before computing correlations.
:returns: The generated Matplotlib figure.
"""
if cmap is None:
cmap = sns.diverging_palette(240, 10, as_cmap=True)
grouping_vars = ['number_of_active_genes', 'number_of_control_genes', 'avg_reads_per_gene',
'classifier_accuracy', 'nr_plates', 'number_of_genes', 'avg_genes_per_well',
'avg_cells_per_well', 'sequencing_error', 'well_ineq_coeff', 'gene_ineq_coeff']
grouping_vars = grouping_vars + ['optimal_threshold', 'accuracy', 'prauc', 'roc_auc','genes_per_well_gini', 'wells_per_gene_gini']
if clean:
df = remove_constant_columns(df)
grouping_vars = [feature for feature in grouping_vars if feature in df.columns]
relevant_data = df[grouping_vars]
if clean:
relevant_data = remove_columns_with_single_value(relevant_data)
corr_matrix = relevant_data.corr()
mask = np.triu(np.ones_like(corr_matrix, dtype=bool))
with figure_style(theme_target(), frame='box'):
fig = plt.figure(figsize=(12, 8))
from .figures.bundle import _register_figure_data
_register_figure_data(fig, corr_matrix, kind="heatmap", matrix=True)
axis = sns.heatmap(corr_matrix, mask=mask, annot=annot, cmap=cmap,
fmt=".2f", linewidths=0, robust=True,
vmin=-1.0, vmax=1.0, center=0.0)
_style_colour_bar(fig)
plt.tight_layout()
plt.show()
save_plot(fig, src=_figures_dst(dst), variable='correlation_matrix', i=1)
return fig
[docs]
def plot_feature_importance(df, target='prauc', exclude=None, clean=True, dst=None):
"""Train a RandomForestRegressor on sweep variables and plot the resulting importances.
:param df: DataFrame with sweep columns and ``target``.
:param target: Column predicted by the regressor. Default ``'prauc'``.
:param exclude: Column name or list of columns to remove from the feature set.
:param clean: When True, drop constant columns before fitting.
:returns: The generated Matplotlib figure.
"""
features = ['number_of_active_genes', 'number_of_control_genes', 'avg_reads_per_gene',
'classifier_accuracy', 'nr_plates', 'number_of_genes', 'avg_genes_per_well',
'avg_cells_per_well', 'sequencing_error', 'well_ineq_coeff', 'gene_ineq_coeff']
if clean:
df = remove_columns_with_single_value(df)
features = [feature for feature in features if feature in df.columns]
if isinstance(exclude, list):
features = [feature for feature in features if feature not in exclude]
elif exclude is not None:
features = [feature for feature in features if feature != exclude]
model = RandomForestRegressor(n_estimators=1000, random_state=42)
model.fit(df[features], df[target])
importances = model.feature_importances_
indices = np.argsort(importances)[::-1]
with figure_style(theme_target()):
fig = plt.figure(figsize=(12, 6))
from .figures.bundle import _register_figure_data
_register_figure_data(fig, lambda: pd.DataFrame({"feature": [str(f) for f in features], "importance": np.asarray(importances, dtype=float)}), x="feature", y="importance", kind="bar")
plt.barh(range(len(indices)), importances[indices],
color=Palette.GREY_DARK, align="center")
plt.yticks(range(len(indices)), [features[i] for i in indices])
plt.gca().invert_yaxis()
plt.xlabel('Feature Importance')
plt.title('Feature Importances')
plt.tight_layout()
plt.show()
save_plot(fig, src=_figures_dst(dst), variable='feature_importance', i=1)
return fig
[docs]
def calculate_permutation_importance(df, target='prauc', exclude=None, n_repeats=10, clean=True, dst=None):
"""Fit a RandomForest and plot permutation-based feature importances for the sweep columns.
:param df: DataFrame with sweep columns and ``target``.
:param target: Column predicted by the regressor. Default ``'prauc'``.
:param exclude: Column name or list of columns to remove from the feature set.
:param n_repeats: Number of permutations per feature. Default ``10``.
:param clean: When True, drop constant columns before fitting.
:returns: The generated Matplotlib figure.
"""
features = ['number_of_active_genes', 'number_of_control_genes', 'avg_reads_per_gene',
'classifier_accuracy', 'nr_plates', 'number_of_genes', 'avg_genes_per_well',
'avg_cells_per_well', 'sequencing_error', 'well_ineq_coeff', 'gene_ineq_coeff']
if clean:
df = remove_columns_with_single_value(df)
features = [feature for feature in features if feature in df.columns]
if isinstance(exclude, list):
features = [feature for feature in features if feature not in exclude]
elif exclude is not None:
features = [feature for feature in features if feature != exclude]
X = df[features]
y = df[target]
model = RandomForestRegressor(n_estimators=100, random_state=42)
model.fit(X, y)
perm_importance = permutation_importance(model, X, y, n_repeats=n_repeats, random_state=42)
sorted_idx = perm_importance.importances_mean.argsort()
with figure_style(theme_target()):
fig, ax = plt.subplots()
from .figures.bundle import _register_figure_data
_register_figure_data(fig, lambda: pd.DataFrame({"feature": [str(f) for f in features], "importance": np.asarray(perm_importance.importances_mean, dtype=float)}), x="feature", y="importance", kind="bar")
ax.barh(range(len(sorted_idx)), perm_importance.importances_mean[sorted_idx],
color=Palette.GREY_DARK, align="center")
ax.set_yticks(range(len(sorted_idx)))
ax.set_yticklabels([features[i] for i in sorted_idx])
ax.set_xlabel('Permutation Importance')
plt.tight_layout()
plt.show()
save_plot(fig, src=_figures_dst(dst), variable='permutation_importance', i=1)
return fig
[docs]
def plot_partial_dependences(df, target='prauc', clean=True, dst=None):
"""Fit a GradientBoostingRegressor and plot partial dependences for every sweep feature.
:param df: DataFrame with sweep columns and ``target``.
:param target: Column predicted by the regressor. Default ``'prauc'``.
:param clean: When True, drop constant columns before fitting.
:returns: The generated Matplotlib figure.
"""
features = ['number_of_active_genes', 'number_of_control_genes', 'avg_reads_per_gene',
'classifier_accuracy', 'nr_plates', 'number_of_genes', 'avg_genes_per_well',
'avg_cells_per_well', 'sequencing_error', 'well_ineq_coeff', 'gene_ineq_coeff']
if clean:
df = remove_columns_with_single_value(df)
features = [feature for feature in features if feature in df.columns]
X = df[features].astype(float)
y = df[target]
model = GradientBoostingRegressor()
model.fit(X, y)
n_cols = 4
n_rows = (len(features) + n_cols - 1) // n_cols
with figure_style(theme_target()):
fig, axs = plt.subplots(nrows=n_rows, ncols=n_cols, figsize=(5 * n_cols, 5 * n_rows))
from .figures.bundle import _register_figure_data
_register_figure_data(fig, df, y=target, kind="line")
fig.suptitle('Partial Dependence Plots',
fontsize=TYPE_SCALE['panel_letter'],
color=resolve_ink(theme_target()), y=1.03)
axs = np.atleast_1d(axs).flatten()
for i, feature in enumerate(features):
ax = axs[i]
PartialDependenceDisplay.from_estimator(
model, X, features=[feature], ax=ax
)
ax.set_title(feature)
for ax in axs[len(features):]:
ax.set_visible(False)
plt.tight_layout()
plt.show()
save_plot(fig, src=_figures_dst(dst), variable='partial_dependences', i=1)
return fig
[docs]
def save_shap_plot(fig, src, variable, i):
"""Save a SHAP figure to ``<src>/<variable>/<i>_figure``.
Through :func:`spacr.plot.save_figure` for the same reason
:func:`save_plot` is: the format follows the preference, not a literal.
:param fig: Matplotlib figure to save.
:param src: base output directory.
:param variable: subdirectory name identifying the plotted variable.
:param i: identifier prefixed to the output filename.
"""
import os
os.makedirs(f'{src}/{variable}', exist_ok=True)
filename_fig = f'{src}/{variable}/{str(i)}_figure.pdf'
filename_fig = save_figure(fig, filename_fig, bbox_inches='tight')
print(f"Saved figure as {filename_fig}")
[docs]
def generate_shap_summary_plot(df,target='prauc', clean=True, dst=None):
"""Fit a RandomForest and render a SHAP summary plot over the standard sweep features.
:param df: DataFrame with sweep columns and ``target``.
:param target: Column predicted by the regressor. Default ``'prauc'``.
:param clean: When True, drop constant columns before fitting.
:returns: The current Matplotlib figure (SHAP creates it as a side effect).
"""
import shap
features = ['number_of_active_genes', 'number_of_control_genes', 'avg_reads_per_gene',
'classifier_accuracy', 'nr_plates', 'number_of_genes', 'avg_genes_per_well',
'avg_cells_per_well', 'sequencing_error', 'well_ineq_coeff', 'gene_ineq_coeff']
if clean:
df = remove_columns_with_single_value(df)
features = [feature for feature in features if feature in df.columns]
X = df[features]
y = df[target]
model = RandomForestRegressor(n_estimators=100, random_state=42)
model.fit(X, y)
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X)
with figure_style(theme_target()):
shap.summary_plot(
shap_values, X, rng=np.random.default_rng(42))
save_shap_plot(plt.gcf(), src=_figures_dst(dst), variable='shap', i=1)
return plt.gcf()
[docs]
def remove_constant_columns(df):
"""Return ``df`` limited to columns that contain more than one unique value.
:param df: Source DataFrame.
:returns: Copy of ``df`` with constant columns dropped.
"""
return df.loc[:, df.nunique() > 1]