"""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()