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