scripts/analysis/99_en1_dual_role_analysis.py

"""En1 dual-role on Dingwall: test A local activator (En1+ > En1- Sweat_gland in WT); test B spatial repressor (cKO > WT Sweat_gland per class)."""
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 mannwhitneyu
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))

# Sweat_gland pathway module (from restored 57_pathway_analysis.py)
SWEAT_GLAND_GENES = ["Foxi3", "Foxa1", "En1", "Krt8", "Krt18", "Krt19",
                     "Muc5b", "Aqp5", "Cutl1"]
# for TEST A we EXCLUDE En1 from the module to avoid tautology
SWEAT_GLAND_MINUS_EN1 = [g for g in SWEAT_GLAND_GENES if g != "En1"]

CKO_GSMS = {"GSM6833482", "GSM6833483"}  # CORRECTED: 480/481 are rttaControl (WT), not cKO
WT_GSMS  = {"GSM6833478", "GSM6833479", "GSM6833480", "GSM6833481"}  # CORRECTED: 4 Cre-neg controls per GEO metadata


def main():
    print("[en1] loading Dingwall raw + v3 marker predictions", flush=True)
    raw = ad.read_h5ad(ROOT / "data/raw/GSE220977_combined.h5ad")
    pred = pd.read_csv(ROOT / "discovery/pan_skin/marker/dingwall_predictions.csv")
    pred_map = dict(zip(pred["cell_id"].astype(str), pred["pred_label"]))
    labels = np.array([pred_map.get(c, "unknown") for c in raw.obs_names.astype(str)])
    raw.obs["pred_label"] = pd.Categorical(labels)
    raw.obs["genotype"] = np.where(raw.obs["sample"].astype(str).isin(list(CKO_GSMS)), "En1-cKO",
                          np.where(raw.obs["sample"].astype(str).isin(list(WT_GSMS)), "WT", "other"))
    labeled = raw.obs["genotype"].isin(["WT", "En1-cKO"]).values
    raw = raw[labeled].copy()
    print(f"[en1] {raw.n_obs:,} labeled cells (WT + cKO); classes: {raw.obs['pred_label'].nunique()}", flush=True)

    sc.pp.normalize_total(raw, target_sum=1e4); sc.pp.log1p(raw)

    # En1 excluded from module to avoid tautology in Test A
    sg_present = [g for g in SWEAT_GLAND_MINUS_EN1 if g in raw.var_names]
    print(f"[en1] Sweat_gland (En1-excluded) genes present: {sg_present}", flush=True)
    sc.tl.score_genes(raw, gene_list=sg_present, score_name="sg_module_minusEn1",
                      random_state=0, use_raw=False)

    if "En1" in raw.var_names:
        en1_col = raw[:, "En1"].X
        en1_exp = en1_col.toarray().flatten() if sp.issparse(en1_col) else en1_col.flatten()
    else:
        print("[en1] WARNING: En1 not detected in var_names", flush=True)
        en1_exp = np.zeros(raw.n_obs)
    raw.obs["En1_expr"] = en1_exp
    raw.obs["En1_detected"] = en1_exp > 0  # any detection after log-norm

    # =======================================================================
    # TEST A — LOCAL ACTIVATOR (En1+ > En1- within WT)
    # =======================================================================
    print("\n=== TEST A: LOCAL ACTIVATOR (En1+ vs En1- Sweat_gland in WT only) ===",
          flush=True)
    wt_mask = raw.obs["genotype"] == "WT"
    wt = raw[wt_mask].copy()
    print(f"[testA] {wt.n_obs} WT cells; En1+ = {int(wt.obs['En1_detected'].sum())}, "
          f"En1- = {int((~wt.obs['En1_detected']).sum())}", flush=True)
    en1pos_score = wt[wt.obs["En1_detected"]].obs["sg_module_minusEn1"].values
    en1neg_score = wt[~wt.obs["En1_detected"]].obs["sg_module_minusEn1"].values
    if len(en1pos_score) > 5 and len(en1neg_score) > 5:
        U_A, p_A = mannwhitneyu(en1pos_score, en1neg_score, alternative="greater")
        delta_A = float(en1pos_score.mean() - en1neg_score.mean())
        print(f"[testA] En1+ mean = {en1pos_score.mean():.4f}, En1- mean = {en1neg_score.mean():.4f}",
              flush=True)
        print(f"[testA] delta = {delta_A:+.4f}, MannU-greater p = {p_A:.3e}", flush=True)
        testA = {
            "n_en1pos": int(len(en1pos_score)), "n_en1neg": int(len(en1neg_score)),
            "mean_en1pos": float(en1pos_score.mean()), "mean_en1neg": float(en1neg_score.mean()),
            "delta_activation": delta_A, "mannu_p_greater": float(p_A),
        }
    else:
        print(f"[testA] insufficient cells; skipping", flush=True)
        testA = {"skipped": True}

    # =======================================================================
    # TEST B — SPATIAL REPRESSOR (cKO > WT in non-eccrine classes)
    # =======================================================================
    print("\n=== TEST B: SPATIAL REPRESSOR (cKO > WT Sweat_gland score per class) ===",
          flush=True)
    rows = []
    for cls in sorted(raw.obs["pred_label"].astype(str).unique()):
        sub = raw[raw.obs["pred_label"].astype(str) == cls]
        wt_cells = sub[sub.obs["genotype"] == "WT"]
        cko_cells = sub[sub.obs["genotype"] == "En1-cKO"]
        if wt_cells.n_obs < 10 or cko_cells.n_obs < 10:
            continue
        wt_scores = wt_cells.obs["sg_module_minusEn1"].values
        cko_scores = cko_cells.obs["sg_module_minusEn1"].values
        U, p_greater = mannwhitneyu(cko_scores, wt_scores, alternative="greater")
        U, p_two = mannwhitneyu(cko_scores, wt_scores, alternative="two-sided")
        delta = float(cko_scores.mean() - wt_scores.mean())
        rows.append({
            "predicted_class": cls,
            "n_WT": int(wt_cells.n_obs),
            "n_cKO": int(cko_cells.n_obs),
            "wt_mean_sg_score": float(wt_scores.mean()),
            "cko_mean_sg_score": float(cko_scores.mean()),
            "delta_derepression": delta,
            "mannu_p_greater_cKO_gt_WT": float(p_greater),
            "mannu_p_two_sided": float(p_two),
            "bonferroni_p_adj": float(min(1.0, p_greater * 20)),  # crude adjustment
            "interpretation": "derepressed_in_cKO" if delta > 0 and p_greater < 0.05
                              else "normal" if delta > 0 else "downregulated_in_cKO",
        })
    df = pd.DataFrame(rows).sort_values("delta_derepression", ascending=False)
    print(df[["predicted_class", "n_WT", "n_cKO", "delta_derepression",
              "mannu_p_greater_cKO_gt_WT", "interpretation"]].to_string(index=False),
          flush=True)

    out = ROOT / "discovery/pan_skin/marker"
    out.mkdir(parents=True, exist_ok=True)
    df.to_csv(out / "99_en1_dual_role.csv", index=False)
    summary = {
        "model": "Dingwall 2024 En1 dual role model",
        "hypothesis": "En1 is BOTH a local activator (turns Sweat_gland ON in eccrine-competent cells) "
                      "AND a spatial repressor (prevents Sweat_gland from firing elsewhere)",
        "test_A_local_activator": testA,
        "test_B_spatial_repressor": {
            "n_classes_tested": len(rows),
            "n_classes_derepressed_p_lt_0.05": int((df["mannu_p_greater_cKO_gt_WT"] < 0.05).sum()),
            "n_classes_derepressed_p_lt_0.001": int((df["mannu_p_greater_cKO_gt_WT"] < 0.001).sum()),
            "top_derepressed": df.head(5)[["predicted_class", "delta_derepression",
                                            "mannu_p_greater_cKO_gt_WT"]].to_dict("records"),
        },
        "module_definition": {
            "name": "Sweat_gland (En1-excluded)",
            "genes_used": sg_present,
            "note": "En1 removed from the module to avoid tautology in Test A "
                    "(En1+ cells trivially score higher on a module containing En1)",
        },
    }
    (out / "99_en1_dual_role_summary.json").write_text(json.dumps(summary, indent=2, default=str))
    print(f"\n[write] {out}/99_en1_dual_role.{{csv,json}}", flush=True)


if __name__ == "__main__":
    main()