scripts/data_prep/compute_celleval_deltas.py

#!/usr/bin/env python
"""Compute cell-eval pseudobulk expression deltas from h5ad files.

For each h5ad file in the input directory:
  1. Load expression matrix (HVG from obsm, or full .X with normalization)
  2. Pseudobulk by (group_col, treatment_col) via sparse matrix multiplication
  3. Delta = mean(treated) - mean(control) per group
  4. Save as wide-format parquet: [group_col, treatment_col, gene1, ..., geneN]

All dataset-specific parameters (column names, control condition, normalization)
are read from a YAML config file. See configs/ for examples.

Usage:
    # Tahoe HVG (2K genes)
    python scripts/data_prep/compute_celleval_deltas.py configs/tahoe.yaml

    # Tahoe full genome (63K genes)
    python scripts/data_prep/compute_celleval_deltas.py configs/tahoe.yaml --profile celleval_full

    # Parse
    python scripts/data_prep/compute_celleval_deltas.py configs/parse.yaml

    # Override output directory
    python scripts/data_prep/compute_celleval_deltas.py configs/tahoe.yaml --output_dir /tmp/test
"""

import argparse
import gc
import glob
from pathlib import Path

import anndata
import numpy as np
import pandas as pd
import scipy.sparse as sp
import yaml
from scipy.sparse import csr_matrix

# ---------------------------------------------------------------------------
# Normalization
# ---------------------------------------------------------------------------


def sparse_normalize_total_log1p(X, target_sum):
    """In-place normalize_total + log1p on a sparse CSR matrix."""
    X = X.tocsr()
    if X.data.dtype.kind != "f":
        X = X.astype(np.float32, copy=False)
    row_sums = np.asarray(X.sum(axis=1)).ravel()
    safe = np.where(row_sums > 0, row_sums, 1.0).astype(X.data.dtype)
    row_idx = np.repeat(np.arange(X.shape[0], dtype=np.int64), np.diff(X.indptr))
    X.data *= (target_sum / safe[row_idx]).astype(X.data.dtype, copy=False)
    np.log1p(X.data, out=X.data)
    return X


def extract_expression(adata, cfg):
    """Extract expression matrix and gene names based on config."""
    gene_mode = cfg.get("gene_mode", "all_genes")

    if gene_mode == "hvg":
        hvg_key = cfg.get("hvg_key", "X_hvg")
        X = np.log1p(adata.obsm[hvg_key])
        if hasattr(X, "toarray"):
            X = X.toarray()
        gene_name_col = cfg.get("gene_name_col", "gene_symbol")
        gene_names = adata.var[gene_name_col].values
    else:
        target_sum = cfg.get("target_sum", 1872)
        if sp.issparse(adata.X):
            X = sparse_normalize_total_log1p(adata.X.copy(), target_sum)
        else:
            row_sums = adata.X.sum(axis=1, keepdims=True)
            row_sums = np.where(row_sums > 0, row_sums, 1.0)
            X = np.log1p(adata.X / row_sums * target_sum)

        gene_name_col = cfg.get("gene_name_col")
        if gene_name_col and gene_name_col in adata.var.columns:
            gene_names = adata.var[gene_name_col].values
        else:
            gene_names = adata.var.index.to_numpy()

    return X, gene_names


# ---------------------------------------------------------------------------
# Pseudobulk
# ---------------------------------------------------------------------------


def pseudobulk_means(X, group_labels):
    """Compute per-group mean expression via sparse group matrix.

    Args:
        X: (n_cells, n_genes) expression matrix (sparse or dense)
        group_labels: (n_cells,) string array of group identifiers

    Returns:
        means: (n_groups, n_genes) ndarray
        group_to_idx: dict mapping group label -> row index in means
    """
    unique_groups, inverse = np.unique(group_labels, return_inverse=True)
    n_groups = len(unique_groups)
    n_cells = len(group_labels)

    counts = np.bincount(inverse, minlength=n_groups)
    weights = (1.0 / counts[inverse]).astype(np.float32)

    group_matrix = csr_matrix(
        (weights, (inverse, np.arange(n_cells))),
        shape=(n_groups, n_cells),
    )

    means = group_matrix @ X
    if sp.issparse(means):
        means = means.toarray()

    group_to_idx = {g: i for i, g in enumerate(unique_groups)}
    return np.asarray(means), group_to_idx


def compute_deltas(means, group_to_idx, gene_names, control, group_col, treatment_col, file_id=None, file_id_col=None):
    """Compute treatment - control deltas.

    Groups are encoded as "group_val||treatment_val" in group_to_idx keys.
    Returns a DataFrame with columns [file_id_col?, group_col, treatment_col, gene1..geneN].
    """
    groups_by_batch = {}
    for key in group_to_idx:
        batch_val, treat_val = key.split("||", 1)
        groups_by_batch.setdefault(batch_val, []).append(treat_val)

    meta_rows = []
    delta_blocks = []
    n_skipped = 0

    for batch_val, treatments in groups_by_batch.items():
        ctrl_key = f"{batch_val}||{control}"
        if ctrl_key not in group_to_idx:
            n_skipped += len([t for t in treatments if t != control])
            continue

        ctrl_mean = means[group_to_idx[ctrl_key]]

        for treat_val in treatments:
            if treat_val == control:
                continue
            treat_key = f"{batch_val}||{treat_val}"
            delta = means[group_to_idx[treat_key]] - ctrl_mean

            row = {}
            if file_id_col and file_id is not None:
                row[file_id_col] = file_id
            row[group_col] = batch_val
            row[treatment_col] = treat_val
            meta_rows.append(row)
            delta_blocks.append(delta)

    if n_skipped > 0:
        print(f"  skipped {n_skipped} condition(s) (no control)")

    if not delta_blocks:
        return pd.DataFrame()

    delta_arr = np.asarray(delta_blocks, dtype=np.float32)
    meta_df = pd.DataFrame(meta_rows)
    gene_df = pd.DataFrame(delta_arr, columns=list(gene_names))
    return pd.concat([meta_df, gene_df], axis=1)


# ---------------------------------------------------------------------------
# Per-file processing
# ---------------------------------------------------------------------------


def process_file(h5ad_path, cfg, global_cfg):
    """Process one h5ad file and return a delta DataFrame."""
    file_stem = Path(h5ad_path).name.split(".h5ad")[0]

    print(f"\n{'=' * 60}")
    print(f"Processing {file_stem}")
    print(f"{'=' * 60}")

    print(f"  Loading {h5ad_path}...")
    adata = anndata.read_h5ad(h5ad_path)
    print(f"  Shape: {adata.n_obs:,} cells x {adata.n_vars:,} genes")

    # Reset var index if needed for gene_name_col access
    if cfg.get("gene_mode") == "all_genes":
        adata.var = adata.var.reset_index()

    X, gene_names = extract_expression(adata, cfg)
    print(f"  Expression: {X.shape[1]:,} genes, mode={cfg.get('gene_mode', 'all_genes')}")

    group_col = global_cfg["group_col"]
    treatment_col = global_cfg["treatment_col"]

    # Build compound group labels: "group_val||treatment_val"
    group_vals = adata.obs[group_col].astype(str).values
    treat_vals = adata.obs[treatment_col].astype(str).values
    labels = np.array([f"{g}||{t}" for g, t in zip(group_vals, treat_vals)])

    means, group_to_idx = pseudobulk_means(X, labels)
    print(f"  Pseudobulk: {len(group_to_idx)} groups")

    file_id_col = global_cfg.get("file_id_col")
    file_id = file_stem if file_id_col else None

    delta_df = compute_deltas(
        means,
        group_to_idx,
        gene_names,
        global_cfg["control"],
        group_col,
        treatment_col,
        file_id=file_id,
        file_id_col=file_id_col,
    )
    print(f"  Output: {delta_df.shape}")

    del adata, X, means
    gc.collect()
    return delta_df


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("config", type=str, help="YAML config file (e.g. configs/tahoe.yaml)")
    parser.add_argument(
        "--profile",
        type=str,
        default="celleval",
        help="Config profile: 'celleval' or 'celleval_full' (default: celleval)",
    )
    parser.add_argument("--output_dir", type=str, default=None, help="Override output directory from config")
    parser.add_argument("--files", nargs="+", default=None, help="Process only these h5ad files (basenames)")
    args = parser.parse_args()

    with open(args.config) as f:
        global_cfg = yaml.safe_load(f)

    profile = global_cfg.get(args.profile, {})
    if not profile:
        raise ValueError(
            f"Profile '{args.profile}' not found in {args.config}. "
            f"Available: {[k for k in global_cfg if isinstance(global_cfg[k], dict)]}"
        )

    output_dir = Path(args.output_dir or profile["output_dir"])
    output_dir.mkdir(parents=True, exist_ok=True)

    h5ad_dir = Path(global_cfg["h5ad_dir"])
    pattern = global_cfg.get("h5ad_pattern", "*.h5ad*")
    h5ad_files = sorted(glob.glob(str(h5ad_dir / pattern)))

    if args.files:
        requested = set(args.files)
        h5ad_files = [f for f in h5ad_files if Path(f).name in requested]

    print(f"Dataset: {global_cfg['dataset']}")
    print(f"Profile: {args.profile}")
    print(f"Gene mode: {profile.get('gene_mode', 'all_genes')}")
    print(f"h5ad files: {len(h5ad_files)} in {h5ad_dir}")
    print(f"Output: {output_dir}")

    all_dfs = []
    for h5ad_path in h5ad_files:
        file_stem = Path(h5ad_path).name.split(".h5ad")[0]
        out_path = output_dir / f"{file_stem}.parquet"

        if out_path.exists():
            print(f"\n  {file_stem}: already done, skipping")
            all_dfs.append(pd.read_parquet(out_path))
            continue

        df = process_file(h5ad_path, profile, global_cfg)
        if not df.empty:
            df.to_parquet(out_path, index=False)
            print(f"  Saved {out_path}")
            all_dfs.append(df)

    # Combine all files
    if all_dfs:
        combined = pd.concat(all_dfs, ignore_index=True)
        combined_name = f"all_{global_cfg['dataset']}_deltas.parquet"
        combined_path = output_dir / combined_name
        combined.to_parquet(combined_path, index=False)
        print(f"\nCombined: {combined.shape} -> {combined_path}")

    print(f"\n{'=' * 60}")
    print("Summary")
    print(f"{'=' * 60}")
    total = sum(len(df) for df in all_dfs)
    n_genes = (
        all_dfs[0].shape[1]
        - len(
            [
                c
                for c in all_dfs[0].columns
                if c in (global_cfg["group_col"], global_cfg["treatment_col"], global_cfg.get("file_id_col", ""))
            ]
        )
        if all_dfs
        else 0
    )
    print(f"  {total:,} conditions x {n_genes:,} genes across {len(all_dfs)} files")


if __name__ == "__main__":
    main()