src/workflows/microarray.py

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

import re
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 os
    import subprocess
    import tempfile

    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{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"]]


# ---------------------------------------------------------------------------
# h5ad feature-ID normalisation
# ---------------------------------------------------------------------------

# var columns that conventionally carry the HGNC symbol, in priority order.
SYMBOL_VAR_CANDIDATES: tuple[str, ...] = (
    "SYMBOL",
    "hgnc_symbol",
    "gene_symbol",
    "symbol",
    "Gene Symbol",
    "gene_name",
)

# Ensembl gene/transcript IDs, with or without a version suffix
# (ENSG00000000003.14). Also matches the mouse/other-species ENSMUSG… form.
_ENSEMBL_RE = re.compile(r"^ENS[A-Z]*[GT]\d+(\.\d+)?$")


def looks_like_gene_symbols(values, sample_size: int = 200) -> bool:
    """
    Heuristic: are these HGNC gene symbols rather than Ensembl/probe IDs?

    Activity scoring matches network target genes by NAME, so a matrix indexed
    by Ensembl IDs silently scores nothing — decoupleR reports zero overlap and
    the whole analysis dies (observed live on ``gse205154_sears``, whose
    ``var.index`` is versioned Ensembl while ``var['SYMBOL']`` holds the symbol).
    """
    vals = [str(v) for v in list(values)[:sample_size] if str(v) not in ("", "nan", "None")]
    if not vals:
        return False
    ensembl = sum(1 for v in vals if _ENSEMBL_RE.match(v))
    return (ensembl / len(vals)) < 0.5


def resolve_symbols_from_var(
    expression_df: pd.DataFrame,
    var: pd.DataFrame,
    collapse_method: str = "mean",
) -> dict[str, Any]:
    """
    Relabel a samples × features matrix from Ensembl/probe IDs to gene symbols.

    Uses the first populated column of ``var`` from ``SYMBOL_VAR_CANDIDATES``.
    Features with no symbol are dropped; duplicate symbols are collapsed with
    ``collapse_duplicate_genes``.

    A no-op (``renamed=False``) when the index already looks like gene symbols
    or no symbol column is available, so it is safe to call unconditionally.

    Returns
    -------
    dict with keys: dataframe, renamed, symbol_column, n_features_before,
    n_features_after, n_unmapped_dropped, n_duplicates_collapsed, warnings.
    """
    out: dict[str, Any] = {
        "dataframe": expression_df,
        "renamed": False,
        "symbol_column": None,
        "n_features_before": int(expression_df.shape[1]),
        "n_features_after": int(expression_df.shape[1]),
        "n_unmapped_dropped": 0,
        "n_duplicates_collapsed": 0,
        "warnings": [],
    }

    if looks_like_gene_symbols(expression_df.columns):
        return out

    if var is None or getattr(var, "empty", True):
        out["warnings"].append(
            "Feature IDs do not look like gene symbols and no var table was "
            "available to map them. Activity scoring matches network genes by "
            "symbol and will find zero overlap."
        )
        return out

    col = next(
        (
            c
            for c in SYMBOL_VAR_CANDIDATES
            if c in var.columns and looks_like_gene_symbols(var[c].dropna())
        ),
        None,
    )
    if col is None:
        out["warnings"].append(
            "Feature IDs do not look like gene symbols and no usable symbol "
            f"column was found in var (looked for {list(SYMBOL_VAR_CANDIDATES)}). "
            "Activity scoring will find zero overlap."
        )
        return out

    symbols = var[col].reindex(expression_df.columns)
    valid = symbols.notna() & ~symbols.astype(str).isin(["", "nan", "None", "-"])
    n_dropped = int((~valid).sum())

    df = expression_df.loc[:, valid.to_numpy()].copy()
    df.columns = symbols[valid].astype(str).to_numpy()

    collapsed = collapse_duplicate_genes(df, method=collapse_method)
    df = collapsed["dataframe"]

    out.update(
        {
            "dataframe": df,
            "renamed": True,
            "symbol_column": col,
            "n_features_after": int(df.shape[1]),
            "n_unmapped_dropped": n_dropped,
            "n_duplicates_collapsed": int(collapsed["n_duplicated_genes"]),
        }
    )
    out["warnings"].append(
        f"Feature IDs were not gene symbols; relabelled from var['{col}'] "
        f"({out['n_features_before']} → {out['n_features_after']} features"
        + (f", {n_dropped} unmapped dropped" if n_dropped else "")
        + (
            f", {collapsed['n_duplicated_genes']} duplicate symbols collapsed "
            f"with method='{collapse_method}'"
            if collapsed["n_duplicated_genes"]
            else ""
        )
        + ")."
    )
    return out