scripts/analysis/105_primary_eden_full_dermal.py

"""primary EDEN discovery on the full dingwall-defined dermal set (not the panda-v3 fibroblast subset)."""
from pathlib import Path
import warnings, json, sys, numpy as np, pandas as pd, anndata as ad, scanpy as sc, scipy.sparse as sp
from scipy.stats import fisher_exact
warnings.filterwarnings("ignore"); sc.settings.verbosity = 0

import os as _os
from pathlib import Path as _Path
PANDA_ROOT = _Path(_os.environ.get("PANDA_ROOT", str(_Path(__file__).resolve().parents[2])))
ROOT = Path(str(PANDA_ROOT))
REPLICA = ROOT / "data/processed/dingwall_replica/dingwall_replica.h5ad"
DERM_MARKERS = ROOT / "data/external_labels/dingwall_supp/biorxiv_media-3.xlsx"
TOP_N = 30

# EDEN identity map from Dingwall paper + Data S2 CellChat
EDEN_IDENTITY = {
    10: "Secondary_EDEN_(Dingwall_cluster_20)",
    2:  "Primary_EDEN_candidate_1_(Derm2_-_immediate_precursor)",
    9:  "Primary_EDEN_candidate_2_(Derm9)",
    6:  "EDEN-signalling_(Derm6)",
    3:  "EDEN-signalling_(Derm3)",
}


def main():
    print("[eden] loading replica dermal set (Dingwall-defined)", flush=True)
    a = ad.read_h5ad(REPLICA)
    # keep only cells the Seurat replica classified as belonging to Dingwall's dermal clusters
    if "is_dermal_paper" in a.obs.columns:
        dermal = a[a.obs["is_dermal_paper"] == True].copy()
    elif "derm_label" in a.obs.columns:
        dermal = a[a.obs["derm_label"] != "non_dermal"].copy()
    else:
        raise RuntimeError("no dermal indicator in replica")
    print(f"[eden] Seurat replica dermal cells: {dermal.n_obs}", flush=True)

    # verify our Derm-label distribution matches the replica
    if "derm_label" in dermal.obs.columns:
        print(f"[eden] Derm label distribution (from replica):", flush=True)
        for k, v in dermal.obs["derm_label"].value_counts().sort_index().items():
            print(f"    {k}: {v}", flush=True)

    # load Data S1C panels
    print(f"\n[eden] loading Data S1C marker panels", flush=True)
    df = pd.read_excel(DERM_MARKERS)
    df = df.sort_values(["cluster", "avg_log2FC"], ascending=[True, False])
    panels = {}
    for cl in sorted(df["cluster"].unique()):
        genes = df[df["cluster"] == cl].head(TOP_N)["gene"].astype(str).tolist()
        panels[int(cl)] = genes

    # replica may or may not have log1p applied; reset from counts layer if present
    if "counts" in dermal.layers:
        dermal.X = dermal.layers["counts"]
    if dermal.X.max() > 30:  # raw counts
        sc.pp.normalize_total(dermal, target_sum=1e4); sc.pp.log1p(dermal)

    # score each cell on all 12 Derm identity panels
    print(f"\n[eden] scoring cells on all 12 Derm panels (top-30 markers each)", flush=True)
    for cl, genes in panels.items():
        present = [g for g in genes if g in dermal.var_names]
        if len(present) < 3:
            dermal.obs[f"derm{cl}_score"] = 0.0
            continue
        sc.tl.score_genes(dermal, gene_list=present, score_name=f"derm{cl}_score",
                          random_state=0, use_raw=False)

    # global baseline
    n_wt = int((dermal.obs["genotype"] == "WT").sum())
    n_cko = int((dermal.obs["genotype"] == "En1-cKO").sum())
    baseline = n_cko / max(n_wt + n_cko, 1)
    print(f"\n[eden] baseline: WT={n_wt} cKO={n_cko} (baseline cKO frac = {baseline:.3f})", flush=True)

    # use the replica's derm_label directly, not argmax of scores
    rows = []
    print(f"\n[eden] per-Derm Fisher exact on replica-assigned identities:", flush=True)
    for cl in sorted(panels.keys()):
        derm_label = f"Derm{cl}"
        if derm_label not in dermal.obs["derm_label"].values:
            continue
        sub = dermal[dermal.obs["derm_label"] == derm_label]
        n_wt_c = int((sub.obs["genotype"] == "WT").sum())
        n_cko_c = int((sub.obs["genotype"] == "En1-cKO").sum())
        if n_wt_c + n_cko_c == 0:
            continue
        cko_frac = n_cko_c / (n_wt_c + n_cko_c)
        n_wt_else = n_wt - n_wt_c
        n_cko_else = n_cko - n_cko_c
        try:
            odds, p_f = fisher_exact([[n_wt_c, n_wt_else], [n_cko_c, n_cko_else]],
                                     alternative="two-sided")
        except ValueError:
            odds, p_f = 1.0, 1.0
        rows.append({
            "derm_id": cl,
            "identity": EDEN_IDENTITY.get(cl, "other"),
            "n_cells": n_wt_c + n_cko_c,
            "n_WT": n_wt_c, "n_cKO": n_cko_c,
            "cko_frac": cko_frac,
            "baseline_cko_frac": baseline,
            "cko_delta": cko_frac - baseline,
            "wt_enrichment_odds_ratio": float(1.0/odds) if odds > 0 else None,
            "fisher_p_two_sided": float(p_f),
            "depletion_direction": "cKO-depleted" if cko_frac < baseline else "cKO-enriched",
            "top10_markers_dingwall_S1C": ", ".join(panels[cl][:10]),
        })

    result_df = pd.DataFrame(rows).sort_values("cko_delta")
    out = ROOT / "discovery/pan_skin/marker"
    out.mkdir(parents=True, exist_ok=True)
    result_df.to_csv(out / "105_primary_eden_full_dermal.csv", index=False)

    print(f"\n{'Derm':<8}{'Identity':<50}{'n':<7}{'WT':<6}{'cKO':<6}{'cKO_frac':<10}"
          f"{'OR (WT enrich)':<16}{'Fisher p':<12}", flush=True)
    print("-" * 130, flush=True)
    for _, r in result_df.iterrows():
        print(f"Derm{r['derm_id']:<5}{r['identity'][:47]:<50}{r['n_cells']:<7}"
              f"{r['n_WT']:<6}{r['n_cKO']:<6}{r['cko_frac']:<10.3f}"
              f"{r['wt_enrichment_odds_ratio']:<16.2f}{r['fisher_p_two_sided']:<12.2e}", flush=True)

    # summary json
    summary = {
        "target": "Dingwall_GSE220977",
        "method": "Seurat-replica-identified 14,251 dermal cells (Dingwall clusters {0,1,3,4,5,8,11,20}); "
                  "per-Derm identity Fisher-exact cKO depletion using replica-assigned Derm labels "
                  "(mapped via Jaccard on top-50 markers to Dingwall Data S1C)",
        "n_dermal_cells_total": int(dermal.n_obs),
        "baseline_cko_frac": float(baseline),
        "n_WT_dermal": n_wt, "n_cKO_dermal": n_cko,
        "per_derm": rows,
    }
    (out / "105_primary_eden_full_dermal.json").write_text(json.dumps(summary, indent=2, default=str))
    print(f"\n[eden] wrote {out}/105_primary_eden_full_dermal.{{csv,json}}", flush=True)


if __name__ == "__main__":
    main()