""" Metadata validation for sample metadata DataFrames and AnnData objects. DESIGN PRINCIPLE ---------------- Manifests define column semantics (roles, allowed values, missing-value meanings, reporting rules). Tools compute counts dynamically from loaded data. This module validates semantics; it does not compare against hardcoded expected counts. 1. Manifest-aware validators (plain-dict input, no pandas required) ------------------------------------------------------------------- validate_condition_column Check that a condition column exists and contains only the declared allowed values. validate_metadata_semantics Check obs columns against metadata_columns semantic definitions: roles, allowed values, missing-value handling, encoded/decoded consistency, and technical-column guards. validate_obs Run all obs-level checks; return combined dict. These accept plain Python dicts derived from AnnData.obs so they can be used in lightweight test environments without importing scanpy. adata_obs_info = { "columns": list(adata.obs.columns), "values": {col: adata.obs[col].tolist() for col in adata.obs.columns}, "n_obs": adata.n_obs, } 2. DataFrame-based helpers (pandas input, no AnnData or scanpy required) ----------------------------------------------------------------------- list_valid_groups Survey all columns for usability as grouping variables (unique value counts, missingness, minimum group size). validate_contrast Validate a DE contrast (test vs control) against a metadata DataFrame, with optional sample subsetting via pandas query syntax. apply_subset Apply a pandas query string to a DataFrame and return clear error messages on failure. Typical call order: subset_result = apply_subset(obs_df, "tumor_subtype != ''") contrast_result = validate_contrast( subset_result["dataframe"], group_column="tumor_subtype", test_group="Classical", control_group="Basal", ) """ from __future__ import annotations from typing import Any import pandas as pd def validate_condition_column( obs_columns: list[str], obs_values: dict[str, list], manifest_obs: dict, ) -> dict[str, Any]: """ Check that the condition column declared in the manifest exists in obs and contains only the expected values. Parameters ---------- obs_columns: Column names present in AnnData.obs. obs_values: Mapping of column_name → list of unique values in that column. manifest_obs: The 'obs' section of the dataset manifest dict. Returns ------- dict with keys: valid (bool), column_found (bool), unexpected_values (list), missing_values (list), message (str). """ condition_col = manifest_obs.get("condition_column") expected_values = set(manifest_obs.get("condition_values", [])) if not condition_col: return { "valid": False, "column_found": False, "unexpected_values": [], "missing_values": sorted(expected_values), "message": "manifest does not declare obs.condition_column", } if condition_col not in obs_columns: return { "valid": False, "column_found": False, "unexpected_values": [], "missing_values": sorted(expected_values), "message": f"condition column '{condition_col}' not found in obs", } actual_values = set(obs_values.get(condition_col, [])) if expected_values: unexpected = sorted(actual_values - expected_values) missing = sorted(expected_values - actual_values) valid = len(unexpected) == 0 else: # No expected values declared — column presence is enough unexpected, missing = [], [] valid = True parts = [] if unexpected: parts.append(f"unexpected values: {unexpected}") if missing: parts.append(f"missing values: {missing}") message = "; ".join(parts) if parts else f"condition column '{condition_col}' is valid" return { "valid": valid, "column_found": True, "unexpected_values": unexpected, "missing_values": missing, "message": message, } def validate_metadata_semantics( obs_columns: list[str], obs_values: dict[str, list], metadata_columns: dict, ) -> dict[str, Any]: """ Validate obs data against the manifest's metadata_columns semantic definitions. Checks that declared columns exist, allowed values are present, technical columns are not used for grouping, and encoded/decoded column pairs are consistent. Does not compare against hardcoded expected counts. Parameters ---------- obs_columns: List of column names present in AnnData.obs. obs_values: Mapping of column_name → list of per-sample values. metadata_columns: Dict of column_name → MetadataColumnDef or plain dict (from manifest). Returns ------- dict with keys: valid (bool), checks (dict of per-column results). """ checks: dict[str, Any] = {} for col_name, col_def in metadata_columns.items(): # Support both MetadataColumnDef objects and plain dicts if hasattr(col_def, "role"): role = col_def.role bio_ok = col_def.biological_grouping_allowed allowed = set(col_def.allowed_values) missing = set(col_def.missing_values) source_col = col_def.source_column decoded_col = col_def.decoded_column interp_warn = col_def.interpretation_warning refusal = col_def.refusal_rules else: role = col_def.get("role", "sample_origin") bio_ok = col_def.get("biological_grouping_allowed", True) allowed = set(col_def.get("allowed_values") or []) missing = set(col_def.get("missing_values") or []) source_col = col_def.get("source_column") decoded_col = col_def.get("decoded_column") interp_warn = col_def.get("interpretation_warning", "") refusal = col_def.get("refusal_rules") or [] issues: list[str] = [] warnings: list[str] = [] # Which column to look for in obs (prefer decoded_column) check_col = decoded_col or col_name # 1. Column presence if check_col not in obs_columns: if source_col and source_col in obs_columns: warnings.append( f"Source column '{source_col}' found but decoded column " f"'{check_col}' is missing — numeric decoder may not have run." ) elif source_col: issues.append( f"Neither decoded column '{check_col}' nor source column " f"'{source_col}' found in obs. Available: {obs_columns}" ) else: issues.append( f"Column '{check_col}' not found in obs. " f"Available: {obs_columns}" ) # 2. Allowed-value check (only when column is present and allowed_values declared) if check_col in obs_columns and allowed: actual = set(obs_values.get(check_col, [])) non_missing = actual - missing unexpected = non_missing - allowed if unexpected: warnings.append( f"Unexpected values in '{check_col}': " f"{sorted(str(v) for v in unexpected)}. " f"Declared allowed values: {sorted(allowed)}" ) # 3. Technical-column guard if not bio_ok: warnings.append( f"Column '{col_name}' has biological_grouping_allowed=False " f"(role='{role}'). Do not use for DE or group comparison." ) # 4. Interpretation and refusal rules surfaced as warnings if interp_warn: warnings.append(f"Interpretation: {interp_warn.strip()}") for rule in refusal: warnings.append(f"Refusal rule: {rule.strip()}") checks[col_name] = { "column": check_col, "role": role, "biological_grouping_allowed": bio_ok, "valid": len(issues) == 0, "issues": issues, "warnings": warnings, } all_valid = all(c["valid"] for c in checks.values()) return {"valid": all_valid, "checks": checks} def validate_obs(adata_obs_info: dict, manifest: "dict | Any") -> dict[str, Any]: """ Run obs-level validation checks against the manifest and return a summary. Checks condition column existence and metadata column semantics. Does NOT compare against hardcoded expected counts — counts are computed dynamically by dataset_count_metadata_values. Parameters ---------- adata_obs_info: Dict describing the loaded AnnData.obs: columns (list[str]) — column names in obs values (dict[str, list]) — column → per-sample value list n_obs (int) — total number of observations manifest: Parsed manifest dict or DatasetManifest instance. Returns ------- dict with keys: valid (bool), checks (dict of check_name → result). """ # Resolve manifest to usable dicts if hasattr(manifest, "metadata_columns"): # DatasetManifest instance manifest_obs = { "condition_column": manifest.group_columns[0] if manifest.group_columns else None, "condition_values": [], } metadata_cols = manifest.metadata_columns else: manifest_obs = manifest.get("obs", {}) metadata_cols = manifest.get("metadata_columns", {}) obs_columns = adata_obs_info.get("columns", []) obs_values = adata_obs_info.get("values", {}) checks: dict[str, Any] = {} # Check 1: condition column presence and allowed values if manifest_obs.get("condition_column"): checks["condition_column"] = validate_condition_column( obs_columns, obs_values, manifest_obs ) # Check 2: metadata semantics (replaces validate_sample_counts) if metadata_cols: checks["metadata_semantics"] = validate_metadata_semantics( obs_columns, obs_values, metadata_cols ) all_valid = all(c.get("valid", False) for c in checks.values()) return {"valid": all_valid, "checks": checks} # --------------------------------------------------------------------------- # DataFrame-based helpers # --------------------------------------------------------------------------- def list_valid_groups( metadata_df: "pd.DataFrame", min_count: int = 3, ) -> dict[str, Any]: """ Survey all columns of a metadata DataFrame for usability as grouping variables. A column is considered usable when it has at least 2 distinct non-null values and at least 2 groups each with ≥min_count samples. Columns where every value is unique (likely sample IDs) are flagged explicitly. Parameters ---------- metadata_df: Sample-level metadata DataFrame (e.g. AnnData.obs or a parsed GEO table). Rows are samples; columns are metadata fields. min_count: Minimum number of samples required per group for a column to be considered usable. Default 3 (minimum for most statistical tests). Returns ------- dict with keys: n_samples (int) — number of rows in the DataFrame. n_columns_checked (int) — number of columns surveyed. usable_columns (list[str]) — columns that pass the usability criteria. columns (dict) — per-column details: n_unique, n_missing, missing_rate, value_counts, unique_values_sample, values_truncated, is_usable, usability_reason. """ n_rows = len(metadata_df) columns_info: dict[str, Any] = {} usable_columns: list[str] = [] for col in metadata_df.columns: series = metadata_df[col] n_missing = int(series.isna().sum()) non_null = series.dropna() n_unique = int(non_null.nunique()) value_counts_raw = non_null.value_counts() value_counts = {str(k): int(v) for k, v in value_counts_raw.items()} # Unique values sample — cap at 20 to keep return value manageable unique_vals_sample = sorted([str(v) for v in non_null.unique()[:20]]) values_truncated = n_unique > 20 # Determine usability is_usable = False if n_rows == 0: usability_reason = "DataFrame is empty" elif n_unique == 0: usability_reason = "All values are missing" elif n_unique == 1: val = str(non_null.iloc[0]) usability_reason = ( f"Only one unique value ('{val}') — cannot form a contrast" ) elif n_unique == n_rows and n_rows > 10: usability_reason = ( f"All {n_unique} values are unique — likely a sample ID column, " "not usable for grouping" ) else: qualifying = [v for v, c in value_counts_raw.items() if c >= min_count] small = [str(v) for v, c in value_counts_raw.items() if c < min_count] if len(qualifying) < 2: usability_reason = ( f"Fewer than 2 groups have ≥{min_count} samples " f"(qualifying groups: {[str(q) for q in qualifying]})" ) else: is_usable = True usability_reason = ( f"{len(qualifying)} group(s) with ≥{min_count} samples" ) if small: usability_reason += ( f"; {len(small)} group(s) below minimum ({small}) " "are present but excluded from contrast" ) columns_info[col] = { "n_unique": n_unique, "n_missing": n_missing, "missing_rate": round(n_missing / n_rows, 4) if n_rows > 0 else 0.0, "value_counts": value_counts, "unique_values_sample": unique_vals_sample, "values_truncated": values_truncated, "is_usable": is_usable, "usability_reason": usability_reason, } if is_usable: usable_columns.append(col) return { "n_samples": n_rows, "n_columns_checked": len(metadata_df.columns), "usable_columns": usable_columns, "columns": columns_info, } def apply_subset( metadata_df: "pd.DataFrame", subset_query: str | None, ) -> dict[str, Any]: """ Apply a pandas query string to a metadata DataFrame. Returns a dict rather than raising so that callers can handle failures gracefully and surface a clear message to the user or agent. Parameters ---------- metadata_df: The metadata DataFrame to filter. subset_query: A pandas query string, e.g. ``"tissue == 'pancreas' and age > 50"``. None or empty string returns the DataFrame unchanged. Returns ------- dict with keys: success (bool) — True if query applied without error. query (str | None) — the query string used. n_before (int) — rows before filtering. n_after (int | None) — rows after filtering; None on failure. n_dropped (int | None) — rows removed; None on failure. dataframe (pd.DataFrame | None) — filtered DataFrame; None on failure. Note: not JSON-serialisable. error (str | None) — error description; None on success. """ n_before = len(metadata_df) if not subset_query or not str(subset_query).strip(): return { "success": True, "query": subset_query, "n_before": n_before, "n_after": n_before, "n_dropped": 0, "dataframe": metadata_df, "error": None, } try: subset = metadata_df.query(subset_query) n_after = len(subset) return { "success": True, "query": subset_query, "n_before": n_before, "n_after": n_after, "n_dropped": n_before - n_after, "dataframe": subset, "error": None, } except Exception as exc: return { "success": False, "query": subset_query, "n_before": n_before, "n_after": None, "n_dropped": None, "dataframe": None, "error": ( f"Query failed: {exc}. " "Use pandas query syntax, e.g. " "\"tumor_subtype == 'Classical'\" or " "\"age > 50 and tissue == 'pancreas'\"." ), } def validate_contrast( metadata_df: "pd.DataFrame", group_column: str, test_group: str, control_group: str, subset_query: str | None = None, min_samples_per_group: int = 3, ) -> dict[str, Any]: """ Validate a differential expression contrast against a metadata DataFrame. Optionally filters the DataFrame first via a pandas query (subset_query) before checking group membership and sample counts. This function ONLY validates metadata and sample selection. It does not run any expression analysis. Use the returned test_sample_ids and control_sample_ids to slice your expression matrix. Parameters ---------- metadata_df: Sample-level metadata DataFrame (e.g. AnnData.obs). Index values are used as sample IDs in the return value. group_column: Column in metadata_df containing the group labels. test_group: Label of the test / foreground group. control_group: Label of the reference / background group. subset_query: Optional pandas query string applied before contrast validation, e.g. ``"tissue == 'tumor'"`` to restrict to tumour samples only. min_samples_per_group: Minimum number of samples required in each group. Default 3. Returns ------- dict with keys: valid (bool) — True if contrast is usable for DE. reason (str | None) — explanation if valid=False, else None. group_column (str) test_group (str) control_group (str) n_test (int | None) — samples in test group; None on early failure. n_control (int | None) — samples in control group; None on early failure. n_total_selected (int | None) — n_test + n_control; None on early failure. test_sample_ids (list[str]) — index values of test-group samples. control_sample_ids (list[str]) — index values of control-group samples. subset_query (str | None) — the subset_query used, if any. warnings (list[str]) — non-fatal issues (e.g. small group sizes, excluded samples). """ warnings: list[str] = [] def _early_return(reason: str) -> dict[str, Any]: return { "valid": False, "reason": reason, "group_column": group_column, "test_group": test_group, "control_group": control_group, "n_test": None, "n_control": None, "n_total_selected": None, "test_sample_ids": [], "control_sample_ids": [], "subset_query": subset_query, "warnings": warnings, } # ── Step 1: Apply optional subset ──────────────────────────────────── working_df = metadata_df if subset_query: sub = apply_subset(metadata_df, subset_query) if not sub["success"]: return _early_return(f"Subset query failed: {sub['error']}") working_df = sub["dataframe"] if len(working_df) == 0: return _early_return( f"Subset query '{subset_query}' selected 0 samples" ) # ── Step 2: group_column existence ─────────────────────────────────── if group_column not in working_df.columns: available = sorted(working_df.columns.tolist()) return _early_return( f"group_column '{group_column}' not found. " f"Available columns: {available}" ) # ── Step 3: Group existence ────────────────────────────────────────── available_groups = working_df[group_column].dropna().unique().tolist() available_str = sorted(str(g) for g in available_groups) if test_group == control_group: return _early_return("test_group and control_group are the same") if test_group not in available_groups: return _early_return( f"test_group '{test_group}' not found in column '{group_column}'. " f"Available groups: {available_str}" ) if control_group not in available_groups: return _early_return( f"control_group '{control_group}' not found in column '{group_column}'. " f"Available groups: {available_str}" ) # ── Step 4: Count and collect sample IDs ───────────────────────────── test_mask = working_df[group_column] == test_group ctrl_mask = working_df[group_column] == control_group test_ids = [str(i) for i in working_df[test_mask].index.tolist()] ctrl_ids = [str(i) for i in working_df[ctrl_mask].index.tolist()] n_test = len(test_ids) n_ctrl = len(ctrl_ids) # ── Step 5: Minimum sample count ───────────────────────────────────── if n_test < min_samples_per_group: return _early_return( f"test_group '{test_group}' has {n_test} sample(s), " f"below minimum {min_samples_per_group}" ) if n_ctrl < min_samples_per_group: return _early_return( f"control_group '{control_group}' has {n_ctrl} sample(s), " f"below minimum {min_samples_per_group}" ) # ── Step 6: Non-fatal warnings ─────────────────────────────────────── if n_test < 5: warnings.append( f"test_group '{test_group}' has only {n_test} samples — " "statistical power may be limited" ) if n_ctrl < 5: warnings.append( f"control_group '{control_group}' has only {n_ctrl} samples — " "statistical power may be limited" ) n_excluded = len(working_df) - n_test - n_ctrl if n_excluded > 0: other = [str(g) for g in available_groups if g not in (test_group, control_group)] warnings.append( f"{n_excluded} sample(s) in other group(s) will be excluded from DE " f"(groups: {other[:5]}{'...' if len(other) > 5 else ''})" ) return { "valid": True, "reason": None, "group_column": group_column, "test_group": test_group, "control_group": control_group, "n_test": n_test, "n_control": n_ctrl, "n_total_selected": n_test + n_ctrl, "test_sample_ids": test_ids, "control_sample_ids": ctrl_ids, "subset_query": subset_query, "warnings": warnings, } def subset_and_require_group( metadata_df: "pd.DataFrame", subset_query: str | None, group_column: str ) -> "pd.DataFrame": """Apply an optional subset query and require group_column to be present. Shared by the activity-stats and microarray workflows. Raises ValueError on a failed or empty query, or a missing column; returns the (possibly subset) frame. """ working_meta = metadata_df if subset_query: try: working_meta = metadata_df.query(subset_query) except Exception as exc: raise ValueError( f"subset_query '{subset_query}' failed: {exc}" ) from exc if len(working_meta) == 0: raise ValueError( f"subset_query '{subset_query}' selected 0 samples." ) if group_column not in working_meta.columns: raise ValueError( f"group_column '{group_column}' not found in metadata. " f"Available: {list(working_meta.columns)}" ) return working_meta