""" Microarray-specific preprocessing and analysis workflow helpers. Reusable, dataset-agnostic logic for datasets declared with data_type: microarray_log_expression or microarray_log_ratio in their manifest. All dataset-specific parameters (platform, gene_symbol_column, collapse_method) come from the manifest features block — not hardcoded here. Manifest helpers (implemented) ------------------------------- get_collapse_params Extract probe collapse parameters from a manifest. check_probe_collapse_needed Simple predicate: does this manifest require collapse? recommend_analysis_path Return ("A"|"B", reason_str) for a manifest. Expression loading and alignment --------------------------------- load_expression_matrix Load CSV/TSV as samples × features DataFrame. harmonize_expression_and_metadata Align expression rows to metadata sample IDs. Data characterisation --------------------- detect_log_scale Heuristic: is the matrix likely log-transformed? Gene-level aggregation ---------------------- collapse_duplicate_genes Collapse duplicate gene-name columns in an already gene-symbol-labelled matrix (not probe→gene mapping; for that use decoupler_collapse_probes_to_genes). Statistical analysis -------------------- prepare_gene_level_statistics Welch's t-test with BH FDR on pre-normalised data. Outputs gene, statistic, pvalue, padj, mean_test, mean_control, log2fc_like. NOT for raw counts — do not use DESeq2 here. DE statistical helpers (used by src/tools/rna.py) -------------------------------------------------- run_welch_ttest Vectorized Welch's t-test; returns DESeq2-schema DataFrame. run_limma Limma moderated t-test via rpy2; returns DESeq2-schema DataFrame. """ from __future__ import annotations from typing import Any import numpy as np import pandas as pd from src.workflows.metadata_validation import subset_and_require_group def get_collapse_params(manifest: "dict | object") -> dict[str, Any]: """ Extract probe-to-gene collapse parameters from a manifest. Reads from the new-schema 'feature_mapping' block if present; falls back to the legacy 'features' block so old YAML manifests still work. Parameters ---------- manifest: A parsed manifest dict OR a DatasetManifest dataclass instance. Returns ------- dict with keys: required (bool) — whether collapse is needed. method (str) — "mean", "max", or "most_variable". gene_symbol_column (str|None) — var column with gene symbols, or None. multi_gene_policy (str) — "drop" or "first". """ # Support both DatasetManifest dataclass and raw dict if hasattr(manifest, "feature_mapping"): fm = manifest.feature_mapping or {} else: fm = manifest.get("feature_mapping") or manifest.get("features") or {} return { "required": bool(fm.get("requires_collapse") or fm.get("collapse_required", False)), "method": fm.get("collapse_method", "mean"), "gene_symbol_column": fm.get("gene_symbol_column"), "multi_gene_policy": fm.get("multi_gene_policy", "drop"), } def check_probe_collapse_needed(manifest: dict) -> bool: """Return True if the manifest declares that probe collapse is required.""" return get_collapse_params(manifest)["required"] def recommend_analysis_path(manifest: "dict | object") -> tuple[str, str]: """ Return (path, reason) where path is "A" or "B". "A" = raw integer counts → DESeq2 pipeline. "B" = pre-normalized/log → ttest or limma pipeline. The path is read from the manifest; this function adds an agent-readable reason string describing why and which tool to use. """ # Support DatasetManifest (uses .analysis_path property) and raw dicts if hasattr(manifest, "analysis_path"): path = str(manifest.analysis_path).upper() data_type = getattr(manifest, "data_level", "unknown") else: path = str(manifest.get("analysis_path", "B")).upper() data_type = manifest.get("data_level") or manifest.get("data_type", "unknown") if path == "A": reason = ( "raw integer counts — use decoupler_preprocess_data then " "decoupler_differential_expression(method='deseq2')" ) else: reason = ( f"{data_type} (pre-normalized) — skip decoupler_preprocess_data; " "use decoupler_differential_expression(method='ttest') or method='limma'" ) return path, reason # --------------------------------------------------------------------------- # Expression loading and alignment # --------------------------------------------------------------------------- def load_expression_matrix(expression_path: str) -> dict[str, Any]: """ Load a gene expression matrix from a CSV or TSV file. Expects samples as rows and features/genes as columns, with sample IDs in the first column (used as the DataFrame index). Parameters ---------- expression_path: Local path or http(s)/ftp URL to a .csv, .tsv, or .txt file. Separator is inferred from the file extension: tab for .tsv/.txt, comma for .csv and all others. Returns ------- dict with keys: dataframe (pd.DataFrame) — samples × features, numeric only. n_samples (int) n_features (int) sample_id_sample (list[str]) — first 5 row index values. feature_id_sample (list[str]) — first 5 column names. warnings (list[str]) — non-fatal issues found during loading. Raises ------ FileNotFoundError if expression_path does not exist. """ from pathlib import Path is_url = expression_path.startswith(("http://", "https://", "ftp://")) if not is_url and not Path(expression_path).exists(): raise FileNotFoundError(f"Expression file not found: {expression_path}") # Infer separator from the extension (strip any URL query string first). suffix = Path(expression_path.split("?", 1)[0]).suffix.lower() sep = "\t" if suffix in (".tsv", ".txt") else "," # pandas reads http(s)/ftp URLs natively; pass the raw string so the URL is # not mangled by Path() (which collapses "https://" → "https:/"). df = pd.read_csv(expression_path, sep=sep, index_col=0) warnings: list[str] = [] # Drop non-numeric columns silently with a warning non_numeric = [c for c in df.columns if not pd.api.types.is_numeric_dtype(df[c])] if non_numeric: df = df.drop(columns=non_numeric) warnings.append( f"Dropped {len(non_numeric)} non-numeric column(s): " f"{non_numeric[:5]}{'...' if len(non_numeric) > 5 else ''}" ) n_samples, n_features = df.shape # Orientation heuristic: far more rows than columns suggests transposition if n_features > 0 and n_samples > n_features * 10 and n_samples > 50: warnings.append( f"Matrix has {n_samples} rows and {n_features} columns. " "For microarray data, samples should be rows and genes columns. " "This may be transposed — verify orientation before proceeding." ) return { "dataframe": df, "n_samples": n_samples, "n_features": n_features, "sample_id_sample": list(df.index[:5].astype(str)), "feature_id_sample": list(df.columns[:5].astype(str)), "warnings": warnings, } def harmonize_expression_and_metadata( expression_df: "pd.DataFrame", metadata_df: "pd.DataFrame", sample_id_column: str | None = None, ) -> dict[str, Any]: """ Align expression rows to metadata sample IDs. Finds the intersection of sample IDs between the expression DataFrame index and the metadata DataFrame (either its index or a specified column), drops samples present in only one source, and returns both DataFrames sorted to the same order. Parameters ---------- expression_df: Samples × features expression DataFrame. metadata_df: Sample metadata DataFrame. sample_id_column: Column in metadata_df whose values are sample IDs. If None, the metadata_df index is used as the sample ID source. Returns ------- dict with keys: expression_df (pd.DataFrame) — aligned expression. metadata_df (pd.DataFrame) — aligned metadata. n_aligned_samples (int) n_expression_only (int) — samples in expression but not metadata. n_metadata_only (int) — samples in metadata but not expression. expression_only_samples (list[str]) metadata_only_samples (list[str]) warnings (list[str]) valid (bool) — True if ≥2 aligned samples remain. """ warnings: list[str] = [] # Resolve metadata index if sample_id_column is not None: if sample_id_column not in metadata_df.columns: raise ValueError( f"sample_id_column '{sample_id_column}' not found in metadata. " f"Available columns: {list(metadata_df.columns)}" ) meta_indexed = metadata_df.set_index(sample_id_column) else: meta_indexed = metadata_df expr_ids = set(expression_df.index.astype(str)) meta_ids = set(meta_indexed.index.astype(str)) common = sorted(expr_ids & meta_ids) expr_only = sorted(expr_ids - meta_ids) meta_only = sorted(meta_ids - expr_ids) if expr_only: warnings.append( f"{len(expr_only)} expression sample(s) have no metadata and will be dropped: " f"{expr_only[:5]}{'...' if len(expr_only) > 5 else ''}" ) if meta_only: warnings.append( f"{len(meta_only)} metadata sample(s) have no expression data and will be dropped: " f"{meta_only[:5]}{'...' if len(meta_only) > 5 else ''}" ) if not common: warnings.append( "No common samples found. " "Check that sample IDs in the expression index match those in the metadata." ) aligned_expr = expression_df.loc[ expression_df.index.astype(str).isin(common) ].sort_index() aligned_meta = meta_indexed.loc[ meta_indexed.index.astype(str).isin(common) ].sort_index() return { "expression_df": aligned_expr, "metadata_df": aligned_meta, "n_aligned_samples": len(common), "n_expression_only": len(expr_only), "n_metadata_only": len(meta_only), "expression_only_samples": expr_only[:20], "metadata_only_samples": meta_only[:20], "warnings": warnings, "valid": len(common) >= 2, } # --------------------------------------------------------------------------- # Data characterisation # --------------------------------------------------------------------------- def detect_log_scale(expression_df: "pd.DataFrame") -> dict[str, Any]: """ Heuristic check for whether an expression matrix is likely log-transformed. Uses value range, integrality, and sign to classify the data. This is a diagnostic helper, not a transformation. Always verify the result against the dataset's documented processing. Returns ------- dict with keys: likely_log_scale (bool) — True if heuristics suggest log scale. likely_log2 (bool) — True if specifically log2 is likely. has_negative_values (bool) value_min (float) value_max (float) value_median (float) value_mean (float) fraction_integer (float) — fraction of values that are whole numbers. diagnostic_notes (list[str]) — reasoning behind the classification. warnings (list[str]) — caveats about heuristic reliability. """ flat = expression_df.values.flatten() flat = flat[~( (flat != flat) | # NaN (flat == float("inf")) | (flat == float("-inf")) )] if len(flat) == 0: return { "likely_log_scale": False, "likely_log2": False, "has_negative_values": False, "value_min": None, "value_max": None, "value_median": None, "value_mean": None, "fraction_integer": None, "diagnostic_notes": ["No finite numeric values found in the matrix."], "warnings": ["Cannot determine scale: matrix contains no finite values."], } vmin = float(np.min(flat)) vmax = float(np.max(flat)) vmed = float(np.median(flat)) vmean = float(np.mean(flat)) frac_int = float(np.mean(flat == np.floor(flat))) has_neg = bool(vmin < 0) notes: list[str] = [] caveats: list[str] = [] likely_log = False likely_log2 = False if has_neg: likely_log = True notes.append( f"Negative values present (min={vmin:.3f}) — consistent with " "log-ratio microarray data centred near 0." ) elif frac_int > 0.9 and vmax > 100: likely_log = False notes.append( f"{frac_int:.0%} of values are integers and max={vmax:.0f} — " "consistent with raw integer counts, not log-transformed." ) elif vmax < 30 and frac_int < 0.1: likely_log = True likely_log2 = True notes.append( f"Max value {vmax:.2f} < 30, values are non-integer — " "consistent with log2-normalised microarray expression " "(log2 CPM or log2 intensity typically ranges 4–18)." ) elif vmax < 50: likely_log = True notes.append( f"Max value {vmax:.2f} — plausibly log-transformed, " "but scale is ambiguous (could be log10 or natural log)." ) else: notes.append( f"Max value {vmax:.2f} > 50 with non-integer values — " "may be RPKM, TPM, or another non-log normalised form. " "Verify against the dataset documentation." ) caveats.append( "This is a value-range heuristic. It cannot distinguish log2 from " "log10 or natural log, and can be fooled by outliers or mixed data." ) return { "likely_log_scale": likely_log, "likely_log2": likely_log2, "has_negative_values": has_neg, "value_min": round(vmin, 4), "value_max": round(vmax, 4), "value_median": round(vmed, 4), "value_mean": round(vmean, 4), "fraction_integer": round(frac_int, 4), "diagnostic_notes": notes, "warnings": caveats, } # --------------------------------------------------------------------------- # Gene-level aggregation # --------------------------------------------------------------------------- def collapse_duplicate_genes( expression_df: "pd.DataFrame", method: str = "mean", ) -> dict[str, Any]: """ Collapse duplicate gene-name columns in a gene-symbol-labelled matrix. This function is for matrices whose columns are already gene symbols with some genes appearing more than once (e.g. after imperfect probe annotation). It is NOT a probe-to-gene mapping step — for that, use decoupler_collapse_probes_to_genes via the MCP tool layer. If no duplicate column names are found, the DataFrame is returned unchanged. Parameters ---------- expression_df: Samples × genes DataFrame. Column names must be gene symbols. method: Aggregation method for duplicates. "mean" — average expression across all duplicate columns. "max" — keep the column with the highest mean expression. "most_variable" — keep the column with the highest variance. Returns ------- dict with keys: dataframe (pd.DataFrame) — samples × unique genes. n_features_before (int) n_features_after (int) n_duplicated_genes (int) — gene names appearing >1 time. duplicated_gene_sample (list[str]) — up to 5 examples. method (str) warnings (list[str]) """ if method not in ("mean", "max", "most_variable"): raise ValueError( f"method must be 'mean', 'max', or 'most_variable', got '{method}'" ) n_before = expression_df.shape[1] col_counts = expression_df.columns.value_counts() dup_genes = col_counts[col_counts > 1].index.tolist() n_dup = len(dup_genes) if n_dup == 0: return { "dataframe": expression_df, "n_features_before": n_before, "n_features_after": n_before, "n_duplicated_genes": 0, "duplicated_gene_sample": [], "method": method, "warnings": [ "No duplicate gene names found; DataFrame returned unchanged." ], } w = [ f"{n_dup} gene name(s) appear more than once; collapsing with method='{method}'.", "This collapses duplicate column names only. If columns are still probe IDs, " "use decoupler_collapse_probes_to_genes first.", ] if method == "mean": collapsed = expression_df.T.groupby(level=0).mean().T elif method == "max": # Per gene: keep the column position with the highest mean expression. # Must use positional indexing (iloc) — label-based indexing on a # DataFrame with duplicate column names returns ALL matching columns, # causing a length mismatch when reassigning column names. col_names = expression_df.columns.tolist() means_arr = expression_df.mean(axis=0).values best_pos: dict[str, int] = {} for i, name in enumerate(col_names): if name not in best_pos or means_arr[i] > means_arr[best_pos[name]]: best_pos[name] = i ordered = list(dict.fromkeys(col_names)) # unique, first-seen order collapsed = expression_df.iloc[:, [best_pos[g] for g in ordered]].copy() collapsed.columns = ordered else: # most_variable # Per gene: keep the column position with the highest variance. # Variance (not mean) carries the differential signal. col_names = expression_df.columns.tolist() var_arr = expression_df.var(axis=0).values best_pos = {} for i, name in enumerate(col_names): if name not in best_pos or var_arr[i] > var_arr[best_pos[name]]: best_pos[name] = i ordered = list(dict.fromkeys(col_names)) collapsed = expression_df.iloc[:, [best_pos[g] for g in ordered]].copy() collapsed.columns = ordered return { "dataframe": collapsed, "n_features_before": n_before, "n_features_after": collapsed.shape[1], "n_duplicated_genes": n_dup, "duplicated_gene_sample": dup_genes[:5], "method": method, "warnings": w, } # --------------------------------------------------------------------------- # Statistical analysis # --------------------------------------------------------------------------- def prepare_gene_level_statistics( expression_df: "pd.DataFrame", metadata_df: "pd.DataFrame", group_column: str, test_group: str, control_group: str, subset_query: str | None = None, method: str = "welch_ttest", ) -> dict[str, Any]: """ Compute gene-level differential statistics for pre-normalised microarray data. Uses Welch's t-test (unequal-variance) with Benjamini-Hochberg FDR correction. Outputs one row per gene with statistic, pvalue, padj, group means, and a log2fc_like column that is a true log2 fold-change only if the input matrix is in log2 scale. This function is for normalised microarray-like expression (log-intensity, log-CPM, log-ratio). Do NOT use on raw integer counts — use DESeq2 for those. Parameters ---------- expression_df: Samples × genes DataFrame. Index must match metadata_df index. metadata_df: Sample metadata DataFrame. Index must match expression_df index. group_column: Column in metadata_df containing the group labels. test_group: Label of the foreground / test condition. control_group: Label of the reference / background condition. subset_query: Optional pandas query string applied to metadata_df before grouping, e.g. ``"tissue == 'tumor'"``. method: Statistical method. Currently only "welch_ttest" is supported. Returns ------- dict with keys: dataframe (pd.DataFrame) — genes × stats, sorted by padj. Columns: statistic, pvalue, padj, mean_test, mean_control, log2fc_like. Index name: "gene". n_genes (int) n_test_samples (int) n_control_samples (int) method (str) group_column, test_group, control_group, subset_query significant_genes_05 (int) — genes with padj < 0.05 significant_genes_01 (int) — genes with padj < 0.01 warnings (list[str]) Raises ------ ValueError if group_column is missing, groups are not found, or either group has fewer than 2 samples. """ import warnings as _w from scipy import stats from statsmodels.stats.multitest import multipletests if method != "welch_ttest": raise ValueError(f"method must be 'welch_ttest', got '{method}'") run_warnings = [ "log2fc_like = mean(test) − mean(control). This equals log2 fold-change " "only when input values are in log2 scale. Verify with detect_log_scale().", "Welch's t-test assumes approximately normal distribution within each group. " "For n < 5, treat p-values as approximate.", "Genes with zero variance in either group are assigned statistic=0, pvalue=1.", ] # Subset (optional) + require the group column (shared helper). working_meta = subset_and_require_group(metadata_df, subset_query, group_column) # ── Align expression to (subset) metadata ──────────────────────────── common_idx = expression_df.index.intersection(working_meta.index) if len(common_idx) == 0: raise ValueError( "No common samples between expression index and metadata index " "after subsetting. Check that indices are aligned." ) aligned_expr = expression_df.loc[common_idx] aligned_meta = working_meta.loc[common_idx] # ── Build group masks ──────────────────────────────────────────────── available = aligned_meta[group_column].unique().tolist() if test_group not in available: raise ValueError( f"test_group '{test_group}' not found in '{group_column}'. " f"Available: {sorted(str(g) for g in available)}" ) if control_group not in available: raise ValueError( f"control_group '{control_group}' not found in '{group_column}'. " f"Available: {sorted(str(g) for g in available)}" ) test_mask = aligned_meta[group_column] == test_group ctrl_mask = aligned_meta[group_column] == control_group n_test = int(test_mask.sum()) n_ctrl = int(ctrl_mask.sum()) if n_test < 2: raise ValueError( f"test_group '{test_group}' has only {n_test} sample(s) — " "need at least 2 for Welch's t-test." ) if n_ctrl < 2: raise ValueError( f"control_group '{control_group}' has only {n_ctrl} sample(s) — " "need at least 2 for Welch's t-test." ) if n_test < 5 or n_ctrl < 5: run_warnings.append( f"Small group sizes (test={n_test}, control={n_ctrl}). " "Statistical power is limited; interpret results with caution." ) # ── Vectorised Welch's t-test ──────────────────────────────────────── X_test = aligned_expr[test_mask].values # (n_test, n_genes) X_ctrl = aligned_expr[ctrl_mask].values # (n_ctrl, n_genes) with _w.catch_warnings(): _w.simplefilter("ignore", RuntimeWarning) t_stats, p_vals = stats.ttest_ind(X_test, X_ctrl, axis=0, equal_var=False) # Replace NaN / Inf from zero-variance genes before BH correction t_stats = np.where(np.isfinite(t_stats), t_stats, 0.0) p_vals = np.where(np.isfinite(p_vals), p_vals, 1.0) _, padj, _, _ = multipletests(p_vals, method="fdr_bh") mean_test = X_test.mean(axis=0) mean_ctrl = X_ctrl.mean(axis=0) log2fc_like = mean_test - mean_ctrl genes = aligned_expr.columns.tolist() result_df = pd.DataFrame( { "statistic": t_stats, "pvalue": p_vals, "padj": padj, "mean_test": mean_test, "mean_control": mean_ctrl, "log2fc_like": log2fc_like, }, index=genes, ) result_df.index.name = "gene" result_df = result_df.sort_values("padj") return { "dataframe": result_df, "n_genes": len(genes), "n_test_samples": n_test, "n_control_samples": n_ctrl, "method": method, "group_column": group_column, "test_group": test_group, "control_group": control_group, "subset_query": subset_query, "significant_genes_05": int((padj < 0.05).sum()), "significant_genes_01": int((padj < 0.01).sum()), "warnings": run_warnings, } # --------------------------------------------------------------------------- # Data type classification # --------------------------------------------------------------------------- def classify_expression_data_type(X_flat: "np.ndarray") -> "dict[str, Any]": """ Classify the expression data type from a flattened expression matrix. Heuristics ---------- raw_counts: whole numbers, no negatives, max > 100. log_expression: non-integer, all positive, max < 35 (log2 CPM/TPM scale). log_ratio: has negatives, max < 15, median near 0 (two-color microarray). unknown: none of the above. Parameters ---------- X_flat: 1-D numpy array of all expression values (adata.X.flatten() or similar). Returns ------- dict with keys: is_integer, has_negatives, value_min, value_max, value_mean, value_median, likely_raw_counts, likely_log_expression, likely_log_ratio, likely_log_transformed, data_type (str). """ is_integer = bool(np.all(X_flat == np.floor(X_flat))) has_negatives = bool(np.any(X_flat < 0)) value_max = float(np.max(X_flat)) value_min = float(np.min(X_flat)) value_mean = float(np.mean(X_flat)) value_median = float(np.median(X_flat)) # Raw integer counts: whole numbers, no negatives, typically large values (>100) likely_raw_counts = is_integer and not has_negatives and value_max > 100 # Single-channel log-expression (log2(CPM+1), log2(TPM+1), etc.): # all positive, non-integer, upper range ~4–25 likely_log_expression = (not is_integer) and (not has_negatives) and (value_max < 35) # Log-ratio microarray (two-color or quantile-normalised single-channel): # centered near 0, has negatives, compact range (typically -5 to +5) # mean(test) - mean(ctrl) on this scale IS a log2 fold-change directly likely_log_ratio = has_negatives and (value_max < 15) and (abs(value_median) < 2) # Combined flag — both sub-types go to Path B likely_log_transformed = likely_log_expression or likely_log_ratio if likely_raw_counts: data_type = "raw_counts" elif likely_log_ratio: data_type = "log_ratio_microarray" elif likely_log_expression: data_type = "log_expression" else: data_type = "unknown" return { "is_integer": is_integer, "has_negatives": has_negatives, "value_min": round(value_min, 4), "value_max": round(value_max, 4), "value_mean": round(value_mean, 4), "value_median": round(value_median, 4), "likely_raw_counts": likely_raw_counts, "likely_log_expression": likely_log_expression, "likely_log_ratio": likely_log_ratio, "likely_log_transformed": likely_log_transformed, "data_type": data_type, } def detect_probe_like_features( feature_names: "pd.Index | list[str]", sample_size: int = 50 ) -> bool: """ Heuristic check for whether feature names look like probe IDs rather than gene symbols. Checks the first sample_size feature names against known probe ID patterns: - Affymetrix: starts with digits then underscore (e.g. "1553551_at") - Illumina: ILMN_ prefix (e.g. "ILMN_1234567") - Long numeric-only: 5+ digits (e.g. "3100001") - Generic long probe: length > 12 with underscore (e.g. "A_23_P100001") Returns True if more than 30% of the sampled features match any pattern. Parameters ---------- feature_names: Feature names to check (e.g. adata.var.index). sample_size: Number of features to sample from the start. Returns ------- bool — True if features look like probe IDs. """ sample = pd.Index(feature_names[: min(sample_size, len(feature_names))]).astype(str) if len(sample) == 0: return False probe_like = ( sample.str.match(r"^\d+_") | sample.str.match(r"^ILMN_\d") | sample.str.match(r"^\d{5,}$") | ((sample.str.len() > 12) & sample.str.contains("_")) ) return bool(probe_like.sum() / len(sample) > 0.3) # --------------------------------------------------------------------------- # Statistical helpers for differential expression # --------------------------------------------------------------------------- def run_welch_ttest( X_test: "np.ndarray", X_ctrl: "np.ndarray", genes: "list[str]" ) -> "pd.DataFrame": """ Welch's t-test vectorized across all genes with Benjamini-Hochberg FDR correction. Assumes X is already in log-scale (log2 or similar), so mean(test) - mean(ctrl) approximates log2 fold-change. Genes with zero variance in either group return NaN from scipy; RuntimeWarnings are suppressed and those genes are replaced with stat=0, pvalue=1 before BH correction (treated as non-differentially-expressed). Returns columns: log2FoldChange, stat, pvalue, padj — same schema as DESeq2 output. """ import warnings from scipy import stats from statsmodels.stats.multitest import multipletests with warnings.catch_warnings(): warnings.simplefilter("ignore", RuntimeWarning) t_stats, p_vals = stats.ttest_ind(X_test, X_ctrl, axis=0, equal_var=False) t_stats = np.where(np.isfinite(t_stats), t_stats, 0.0) p_vals = np.where(np.isfinite(p_vals), p_vals, 1.0) log2fc = np.mean(X_test, axis=0) - np.mean(X_ctrl, axis=0) _, padj, _, _ = multipletests(p_vals, method="fdr_bh") return pd.DataFrame( {"log2FoldChange": log2fc, "stat": t_stats, "pvalue": p_vals, "padj": padj}, index=genes, ) def run_limma( X_test: "np.ndarray", X_ctrl: "np.ndarray", genes: "list[str]", test_group: str, control_group: str, ) -> "pd.DataFrame": """ Limma moderated t-test via Rscript subprocess — no rpy2 bridge. After 4 failed rpy2 approaches (deprecated activate(), py2rpy conversion, non-conformable arrays from round-trip, unknown conversion errors), we bypass rpy2 entirely. The expression matrix is written to a temp CSV, limma runs in a fresh Rscript process, and results are read back as CSV. Rscript is installed via packages.txt (r-base) and has been independently verified to work. Raises RuntimeError if Rscript/limma is unavailable — caller falls back to ttest. Returns columns: log2FoldChange, stat, pvalue, padj — same schema as DESeq2. """ import subprocess import tempfile import os n_test, n_ctrl = X_test.shape[0], X_ctrl.shape[0] X_all = np.vstack([X_test, X_ctrl]).T.astype(np.float64) # genes × samples with tempfile.TemporaryDirectory() as tmpdir: expr_csv = os.path.join(tmpdir, "expr.csv") result_csv = os.path.join(tmpdir, "result.csv") # Write expression matrix: genes as rows, samples as columns expr_df = pd.DataFrame(X_all, index=genes) expr_df.to_csv(expr_csv, header=False) r_script = f""" suppressPackageStartupMessages({{ library(limma) library(utils) }}) expr <- as.matrix(read.csv("{expr_csv}", header=FALSE, row.names=1)) n_test <- {n_test} n_ctrl <- {n_ctrl} group <- factor(c(rep("test", n_test), rep("ctrl", n_ctrl)), levels = c("ctrl", "test")) design <- model.matrix(~ group) fit <- lmFit(expr, design) fit <- eBayes(fit) result <- topTable(fit, coef = "grouptest", number = nrow(expr), sort.by = "none", adjust.method = "BH") write.csv(result, "{result_csv}", row.names = TRUE) cat("OK\\n") """ proc = subprocess.run( ["Rscript", "--vanilla", "-"], input=r_script, capture_output=True, text=True, timeout=300, ) if proc.returncode != 0: raise RuntimeError( f"Rscript/limma failed (exit {proc.returncode}):\n{proc.stderr.strip()}" ) if not os.path.exists(result_csv): raise RuntimeError( f"Rscript ran but produced no output.\nstdout: {proc.stdout}\nstderr: {proc.stderr}" ) top_df = pd.read_csv(result_csv, index_col=0) top_df = top_df.rename(columns={ "logFC": "log2FoldChange", "t": "stat", "P.Value": "pvalue", "adj.P.Val": "padj", }) top_df.index = genes return top_df[["log2FoldChange", "stat", "pvalue", "padj"]] def run_limma_covariate( X: "np.ndarray", group_labels: "list[str]", batch_labels: "list[str]", genes: "list[str]", test_group: str, control_group: str, ) -> "pd.DataFrame": """ Limma with a batch covariate via Rscript — ``model.matrix(~ batch + group)``. Mode-A early integration (ADR-0001 T8): when several cohorts are pooled into one matrix, the per-cohort ``batch`` is modelled as a covariate so the group effect is estimated *adjusting* for it. Group labels are recoded to ctrl/test (control_group -> "ctrl", test_group -> "test") so the tested coefficient is always "grouptest" regardless of the original label spelling; batch enters as additional factor columns whose names do not matter (only the group coefficient is read back). Parameters ---------- X : samples x genes matrix (rows = samples, aligned with group/batch labels). group_labels, batch_labels : per-sample labels, length == X.shape[0]. genes : gene ids, length == X.shape[1]. Returns columns: log2FoldChange, stat, pvalue, padj (same schema as run_limma). Raises RuntimeError if Rscript/limma is unavailable or the design is rank- deficient (e.g. batch perfectly confounded with group). """ import os import subprocess import tempfile X = np.asarray(X, dtype=np.float64) n = X.shape[0] if not (len(group_labels) == len(batch_labels) == n): raise ValueError( f"group_labels ({len(group_labels)}) and batch_labels " f"({len(batch_labels)}) must match X sample count ({n})" ) coded = ["test" if str(g) == str(test_group) else "ctrl" for g in group_labels] if len(set(coded)) < 2: raise ValueError( f"need both groups present; got only {set(group_labels)} for " f"test='{test_group}' / control='{control_group}'" ) expr = X.T # genes x samples with tempfile.TemporaryDirectory() as tmpdir: expr_csv = os.path.join(tmpdir, "expr.csv") meta_csv = os.path.join(tmpdir, "meta.csv") result_csv = os.path.join(tmpdir, "result.csv") pd.DataFrame(expr, index=genes).to_csv(expr_csv, header=False) pd.DataFrame( {"group": coded, "batch": [str(b) for b in batch_labels]} ).to_csv(meta_csv, index=False) r_script = f""" suppressPackageStartupMessages({{ library(limma) library(utils) }}) expr <- as.matrix(read.csv("{expr_csv}", header=FALSE, row.names=1)) meta <- read.csv("{meta_csv}", colClasses = "character") group <- factor(meta$group, levels = c("ctrl", "test")) batch <- factor(meta$batch) design <- model.matrix(~ batch + group) if (qr(design)$rank < ncol(design)) {{ stop("design is rank-deficient (batch likely confounded with group)") }} fit <- lmFit(expr, design) fit <- eBayes(fit) result <- topTable(fit, coef = "grouptest", number = nrow(expr), sort.by = "none", adjust.method = "BH") write.csv(result, "{result_csv}", row.names = TRUE) cat("OK\\n") """ proc = subprocess.run( ["Rscript", "--vanilla", "-"], input=r_script, capture_output=True, text=True, timeout=300, ) if proc.returncode != 0: raise RuntimeError( f"Rscript/limma (covariate) failed (exit {proc.returncode}):\n" f"{proc.stderr.strip()}" ) if not os.path.exists(result_csv): raise RuntimeError( f"Rscript ran but produced no output.\nstdout: {proc.stdout}\n" f"stderr: {proc.stderr}" ) top_df = pd.read_csv(result_csv, index_col=0) top_df = top_df.rename(columns={ "logFC": "log2FoldChange", "t": "stat", "P.Value": "pvalue", "adj.P.Val": "padj", }) top_df.index = genes return top_df[["log2FoldChange", "stat", "pvalue", "padj"]]