"""Shared base for the dataset_tools package: imports, dataset_mcp, helpers, intent constants.""" from __future__ import annotations # ruff: noqa: F401, E402 (imports re-exported to tool submodules; intentional late imports) import sys from pathlib import Path from typing import Annotated, Any, Optional import pandas as pd from fastmcp import FastMCP _PROJECT_ROOT = Path(__file__).parent.parent.parent.resolve() if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) from src.datasets.manifest_schema import validate_manifest from src.datasets.registry import get_integration_plan, list_available_datasets, load_manifest from src.workflows.activity_stats import get_contrast_groups from src.workflows.metadata_validation import list_valid_groups, validate_contrast from src.workflows.microarray import get_collapse_params, recommend_analysis_path dataset_mcp = FastMCP(name="dataset") # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _load_metadata_df(metadata_path: str) -> pd.DataFrame: """ Load sample metadata into a DataFrame from .h5ad, .csv, or .tsv. For .h5ad files, returns adata.obs (sample-level metadata only). Raises FileNotFoundError if the path does not exist. """ path = Path(metadata_path) if not path.exists(): raise FileNotFoundError(f"Metadata file not found: {metadata_path}") suffix = path.suffix.lower() if suffix == ".h5ad": import scanpy as sc adata = sc.read_h5ad(str(path)) return adata.obs.copy() elif suffix in (".tsv", ".txt"): return pd.read_csv(str(path), sep="\t", index_col=0) else: return pd.read_csv(str(path), index_col=0) def _build_loading_plan(manifest: Any) -> list[dict]: """Build a step-by-step loading plan from a DatasetManifest.""" source = manifest.expression_source meta_source = manifest.metadata_source collapse = get_collapse_params(manifest) contrast = get_contrast_groups(manifest) pid = manifest.dataset_id steps = [] step = 1 collapse_precomputed = False # Single-cell / spatial (Path P) is decided by MODALITY, not the source-type # string: the expression_source is a hosted AnnData h5ad whether the manifest # declares it as `h5ad` or as `url` pointing at a `.h5ad` (biodata-registry # uses `url`). Load via the sc loader (read_h5ad_cached, ADR-0006 Role 1) — # NOT the bulk flat-file loader `decoupler_load_url_counts`. if manifest.analysis_path == "P": steps.append( { "step": step, "tool": "decoupler_load_and_visualize_data", "key_args": { "adata_path": source.get("url"), "out_prefix": f"{pid}_loaded", }, "note": ( "Single-cell / spatial h5ad (ADR-0006 Role 1). Loads the hosted " "AnnData via read_h5ad_cached — parsed once per resident MCP " "process; a hosted/private h5ad URL is resolved to a local path " "via the authenticated loader (HF_TOKEN). The full integrated " "atlas is NOT served on this live path — score its offline-derived " "signatures against bulk cohorts with dataset_score_signature." ), } ) step += 1 expr_h5ad = f"" elif source.get("type") == "geo_series_matrix": steps.append( { "step": step, "tool": "decoupler_load_geo_series_matrix", "key_args": { "url_or_path": source.get("url"), "condition_column": manifest.group_columns[0] if manifest.group_columns else None, "out_prefix": f"{pid}_raw", }, "note": "Downloads GEO series matrix, decodes numeric condition codes, saves h5ad.", } ) step += 1 expr_h5ad = f"" elif source.get("type") == "url": collapsed_url = source.get("collapsed_url") collapse_precomputed = bool(collapsed_url) and collapse["required"] if collapse_precomputed: load_note = ( f"Downloads precomputed gene-collapsed expression " f"(probes already collapsed via {collapse['method']}); " f"no separate decoupler_collapse_probes_to_genes step needed." ) else: load_note = ( f"Downloads flat genes × samples matrix (URL type). " f"feature_id_type='{manifest.feature_id_type}'. " f"Transposes to AnnData convention (samples × genes)." ) steps.append( { "step": step, "tool": "decoupler_load_url_counts", "key_args": { "url_or_path": collapsed_url if collapse_precomputed else source.get("url"), "feature_id_type": manifest.feature_id_type, "strip_ensembl_versions": manifest.feature_id_type == "ensembl_gene_id", "out_prefix": f"{pid}_raw", }, "note": load_note, } ) step += 1 expr_h5ad = f"" # Clinical join: only when metadata_source has a URL (not embedded) clin_url = meta_source.get("url") if meta_source else None if clin_url and not meta_source.get("embedded", False): steps.append( { "step": step, "tool": "decoupler_join_clinical_metadata", "key_args": { "adata_path": expr_h5ad, "clinical_url_or_path": clin_url, "barcode_column": meta_source.get("join_column"), "truncate_to_patient": meta_source.get("truncate_to_patient", True), "add_survival_columns": bool(manifest.survival_columns.get("event_column")), "out_prefix": f"{pid}_clinical", }, "note": ( "Joins clinical/phenotype TSV to the expression h5ad by sample barcode. " "Derives os_event/os_days from vital_status + days_to_death if survival columns are defined." ), } ) step += 1 expr_h5ad = f"" # Curation step: filter to curated sample list if defined in manifest if getattr(manifest, "curated_sample_list", None): n_curated = len(manifest.curated_sample_list) steps.append( { "step": step, "tool": "dataset_filter_to_curated_samples", "key_args": { "adata_path": expr_h5ad, "dataset_id": pid, "out_prefix": f"{pid}_curated", }, "note": ( f"Filters to {n_curated} curated samples from the manifest " f"({getattr(manifest, 'curated_sample_source', 'see manifest')}). " "Required before survival or prognostic analysis to remove " "non-target tissue contamination." ), } ) step += 1 expr_h5ad = f"" elif source.get("type") == "gdc": steps.append( { "step": step, "tool": "decoupler_load_gdc_star_counts", "key_args": { "project_id": source.get("project_id", "TCGA-PAAD"), "count_column": source.get("count_column", "unstranded"), "feature_id_type": manifest.feature_id_type, "strip_ensembl_versions": manifest.feature_id_type == "ensembl_gene_id", "out_prefix": f"{pid}_raw", }, "note": ( f"Downloads STAR-Counts from GDC API for {source.get('project_id', 'TCGA-PAAD')}. " "Bulk-downloads ~178 per-sample TSVs as a single tar.gz (~500 MB). " "Returns integer raw counts — Path A (DESeq2). No authentication required." ), } ) step += 1 expr_h5ad = f"" # Clinical join: only when metadata_source has a URL (not embedded) clin_url = meta_source.get("url") if meta_source else None if clin_url and not meta_source.get("embedded", False): steps.append( { "step": step, "tool": "decoupler_join_clinical_metadata", "key_args": { "adata_path": expr_h5ad, "clinical_url_or_path": clin_url, "barcode_column": meta_source.get("join_column"), "truncate_to_patient": meta_source.get("truncate_to_patient", True), "add_survival_columns": bool(manifest.survival_columns.get("event_column")), "out_prefix": f"{pid}_clinical", }, "note": ( "Joins clinical/phenotype TSV to the expression h5ad by sample barcode. " "Derives os_event/os_days from vital_status + days_to_death if survival columns are defined." ), } ) step += 1 expr_h5ad = f"" # Curation step: filter to curated sample list if defined in manifest if getattr(manifest, "curated_sample_list", None): n_curated = len(manifest.curated_sample_list) steps.append( { "step": step, "tool": "dataset_filter_to_curated_samples", "key_args": { "adata_path": expr_h5ad, "dataset_id": pid, "out_prefix": f"{pid}_curated", }, "note": ( f"Filters to {n_curated} curated samples from the manifest " f"({getattr(manifest, 'curated_sample_source', 'see manifest')}). " "Required before survival or prognostic analysis to remove " "non-target tissue contamination." ), } ) step += 1 expr_h5ad = f"" elif source.get("type") == "h5ad": # Hosted AnnData .h5ad — single-cell / spatial cohorts (ADR-0006 Role 1, # e.g. the Loveless Steele subset). Loaded once per resident MCP process # via read_h5ad_cached. steps.append( { "step": step, "tool": "decoupler_load_and_visualize_data", "key_args": { "adata_path": source.get("url"), "out_prefix": f"{pid}_loaded", }, "note": ( "Single-cell / spatial h5ad (ADR-0006 Role 1). Loads the hosted " "AnnData via read_h5ad_cached — parsed once per resident MCP " "process; the Steele-subset is MB-scale so copy-on-read is fine. " "A hosted/private h5ad URL is resolved to a local path via the " "authenticated loader (HF_TOKEN). The full integrated atlas is " "NOT served on this live path — score its offline-derived " "signatures against bulk cohorts with dataset_score_signature." ), } ) step += 1 expr_h5ad = f"" else: # Unsupported source type — still emit inspect_data with a warning expr_h5ad = "" steps.append( { "step": step, "tool": "decoupler_inspect_data", "key_args": {"adata_path": expr_h5ad}, "note": ( f"Expected data_type='{manifest.data_level}', " f"analysis_path='{manifest.analysis_path}', " f"features_look_like_probes={collapse['required'] and not collapse_precomputed}." ), } ) step += 1 if collapse["required"] and not collapse_precomputed: steps.append( { "step": step, "tool": "decoupler_collapse_probes_to_genes", "key_args": { "adata_path": expr_h5ad, "gene_symbol_column": collapse.get("gene_symbol_column"), "method": collapse["method"], "out_prefix": f"{pid}_collapsed", }, "note": "Collapses probe IDs to gene symbols before enrichment tools.", } ) de_input = f"" step += 1 else: de_input = expr_h5ad # PATH P (single-cell / spatial): NOT the bulk DESeq2/limma contrast path. # Per-cell work is loaded via read_h5ad_cached; query-time analysis uses the # rna_sc per-cell scoring tools, and any group contrast goes through # pseudobulk aggregation before the bulk DE / activity fast path. if manifest.analysis_path == "P": steps.append( { "step": step, "tool": "decoupler_score_transcription_factors", "key_args": {"adata_path": de_input}, "note": ( "PATH P (single-cell / spatial): score per-cell activity with " "the rna_sc tools (decoupler_score_transcription_factors / " "_progeny_pathways / _hallmark / _cell_types). For a group " "CONTRAST, aggregate to pseudobulk first, then run the bulk DE / " "activity fast path — do NOT run DESeq2/limma on per-cell counts." ), } ) step += 1 return steps # Path A only: filter and preprocess before DESeq2 if manifest.analysis_path == "A": steps.append( { "step": step, "tool": "decoupler_load_and_filter_data", "key_args": { "counts_path": de_input, "group_column": contrast.get("design_factor", "condition"), "out_prefix": f"{pid}_filtered", }, "note": "PATH A: Filter low-expression genes before DESeq2.", } ) step += 1 de_input = f"" steps.append( { "step": step, "tool": "decoupler_preprocess_data", "key_args": { "adata_path": de_input, "out_prefix": f"{pid}_preprocessed", }, "note": "PATH A: Normalize, scale, and PCA before DESeq2.", } ) step += 1 de_input = f"" steps.append( { "step": step, "tool": "dataset_validate_contrast", "key_args": { "metadata_path": de_input, "group_column": contrast.get("design_factor", ""), "test_group": contrast.get("test_group", ""), "control_group": contrast.get("control_group", ""), }, "note": "Confirm groups exist and have enough samples before running DE.", } ) step += 1 de_method = contrast.get("method", "ttest") fallback_method = contrast.get("fallback_method") steps.append( { "step": step, "tool": "decoupler_differential_expression", "key_args": { "adata_path": de_input, "design_factor": contrast.get("design_factor", "condition"), "contrast": [ contrast.get("design_factor", "condition"), contrast.get("test_group", ""), contrast.get("control_group", ""), ], "method": de_method, "fallback_to_ttest": fallback_method == "ttest", "out_prefix": f"{pid}_de", }, "note": ( f"analysis_path='{manifest.analysis_path}': " + ( "use deseq2" if manifest.analysis_path == "A" else f"use {de_method}" + (f" (fallback: {fallback_method})" if fallback_method else "") + ", not deseq2" ) ), } ) return steps # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- # =========================================================================== # Analysis planner # =========================================================================== # --------------------------------------------------------------------------- # Keyword tables (no LLM — pure substring matching on lowercased question) # --------------------------------------------------------------------------- _COMPARE_KW: frozenset[str] = frozenset( { "compare", "comparison", "differ", "difference", "differences", "vs", "versus", "between", "higher in", "lower in", "higher than", "lower than", "upregulated", "downregulated", "up in", "down in", "enriched", "depleted", "differential", "significant", "test vs", "control vs", } ) _SCORE_KW: frozenset[str] = frozenset( { "activity", "activities", "score", "scores", "scoring", "which samples", "sample-level", "per sample", "across samples", "pathway activity", "tf activity", "hallmark activity", "progeny", "collectri", "ranked by", "rank samples", "high tf", "high pathway", "low tf", "low pathway", } ) _SURVIVAL_KW: frozenset[str] = frozenset( { "survival", "prognosis", "prognostic", "overall survival", "disease-free", "recurrence", "mortality", "time to event", "kaplan", "kaplan-meier", "cox regression", "cox model", "hazard ratio", "log-rank", "outcome", "death", "associated with survival", "survival analysis", } ) _CORRELATE_KW: frozenset[str] = frozenset( { "correlate", "correlation", "correlates", "associated with", "association", "covariate", "continuous", "regression", "trend", "predict", "predictor", "relate to", "relation between", } ) _METADATA_KW: frozenset[str] = frozenset( { "how many", "count", "counts", "sample count", "sample counts", "distribution", "group distribution", "subtype distribution", "break down", "breakdown", "value counts", "value_counts", "how many classical", "how many basal", "how many samples", "group sizes", "what groups", "what subtypes", "what categories", "list samples", "describe metadata", "summarise metadata", "summarize metadata", "metadata summary", "column values", "what are the values", "what values", } ) _ALL_KW: dict[str, frozenset[str]] = { "metadata_summary": _METADATA_KW, "compare_groups": _COMPARE_KW, "score_samples": _SCORE_KW, "survival": _SURVIVAL_KW, "correlate_continuous": _CORRELATE_KW, } # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _detect_intent(question: str) -> tuple[str, str, list[str]]: """ Return (intent, confidence, matched_keywords). Uses substring matching on the lowercased question — no LLM. confidence: "high" (≥2 matches), "medium" (1 match), "low" (0 matches). Ties broken by insertion order of _ALL_KW (compare_groups has priority). """ q = question.lower() scores: dict[str, int] = { intent: sum(1 for kw in kws if kw in q) for intent, kws in _ALL_KW.items() } best_intent = max(scores, key=scores.get) best_score = scores[best_intent] if best_score == 0: return "unknown", "low", [] matched = [kw for kw in _ALL_KW[best_intent] if kw in q] confidence = "high" if best_score >= 2 else "medium" return best_intent, confidence, matched # The prefix of _build_loading_plan that is about OPENING the dataset. Anything # else it emits (validate_contrast, DE, scoring) is the analysis tail, which the # intent-specific steps in _workflow_for_intent own instead. _LOADING_PLAN_TOOLS = frozenset( { "decoupler_load_url_counts", "decoupler_load_geo_series_matrix", "decoupler_load_gdc_star_counts", "decoupler_load_and_visualize_data", "decoupler_annotate_probes_with_gpl", "decoupler_collapse_probes_to_genes", "decoupler_join_clinical_metadata", "dataset_filter_to_curated_samples", } ) def _workflow_for_intent( intent: str, dataset_id: str, manifest: Any | None, ) -> dict[str, Any]: """ Build the ordered workflow steps and metadata for a detected intent. All dataset-specific arg values (organism, contrast groups, etc.) are drawn from the manifest when available. Falls back to generic placeholders when no manifest is loaded. """ # Extract manifest hints analysis_path = "B" collapse_required = False contrast: dict[str, str] = {} organism = "human" if manifest is not None: analysis_path = getattr(manifest, "analysis_path", "B") organism = getattr(manifest, "organism", "human") collapse_required = get_collapse_params(manifest)["required"] contrast = get_contrast_groups(manifest) step_n = 1 # Preamble: always describe the dataset first steps: list[dict] = [ { "step": step_n, "tool": "dataset_describe", "purpose": "Confirm data type, analysis path, and default contrast.", "status": "available", "args_hint": {"dataset_id": dataset_id}, } ] step_n += 1 # Loading steps come from _build_loading_plan — the single source of truth # for HOW to open a dataset. This used to emit a load step only for # `geo_series_matrix`, so every h5ad-over-`url` dataset (most of the # registry) got a plan with NO loading step at all and the agent had to # guess an entry point, which cost 3-5 steps per run and often failed # outright. The loading plan also knows about `collapsed_url`, which skips # the annotate+collapse pair below. loading_steps: list[dict] = [] if manifest is not None and hasattr(manifest, "expression_source"): try: loading_steps = _build_loading_plan(manifest) except Exception: # a malformed manifest must not sink the whole plan loading_steps = [] collapse_precomputed = False for raw in loading_steps: tool = raw.get("tool") # Take only the GET-THE-DATA-OPEN prefix. _build_loading_plan also # carries an analysis tail (validate_contrast → DE → scoring); the # intent-specific steps below own that, and splicing both in would # duplicate it. if tool not in _LOADING_PLAN_TOOLS: continue # The loading plan's own inspect step is redundant for a REGISTERED # dataset — data_level and analysis_path are manifest facts (efficiency # rule 4). Keeping it here is what kept the agent calling it. if tool == "decoupler_inspect_data": continue if tool == "decoupler_collapse_probes_to_genes": collapse_precomputed = True steps.append( { "step": step_n, "tool": tool, "purpose": raw.get("note", "Load the dataset as the manifest specifies."), "status": "available", "args_hint": raw.get("key_args", {}), } ) step_n += 1 # Only add a collapse step if the loading plan did not already cover it # (a precomputed `collapsed_url` makes it unnecessary). if collapse_required and not collapse_precomputed and not loading_steps: steps.append( { "step": step_n, "tool": "decoupler_collapse_probes_to_genes", "purpose": "Collapse probe IDs to gene symbols (required before enrichment).", "status": "available", "args_hint": get_collapse_params(manifest) if manifest else {}, } ) step_n += 1 # ── Intent-specific steps ──────────────────────────────────────────────── if intent == "metadata_summary": steps += [ { "step": step_n, "tool": "dataset_interpret_metadata", "purpose": "Show semantic definitions for each metadata column: roles, " "allowed values, missing-value meanings, and analysis rules.", "status": "available", "args_hint": { "adata_path": "", "dataset_id": dataset_id, }, }, { "step": step_n + 1, "tool": "dataset_count_metadata_values", "purpose": "Compute value counts dynamically from the loaded data, " "applying manifest missing-value semantics to classify " "annotated vs unannotated samples.", "status": "available", "args_hint": { "adata_path": "", "dataset_id": dataset_id, }, }, ] return { "steps": steps, "required_inputs": ["h5ad path from decoupler_load_geo_series_matrix"], "assumptions": [ "Dataset has been loaded and saved as h5ad.", "Counts are computed from the loaded data, not from the manifest.", ], "warnings": [ "Sample counts are computed dynamically — they may differ from " "documentation if the dataset is subsetted or filtered.", ], "refusal_conditions": [ "Refuse to report counts as absolute truth without noting the " "data version and any applied subset query.", ], } if intent == "compare_groups": de_method = "ttest" if analysis_path == "B" else "deseq2" steps += [ { "step": step_n, "tool": "dataset_list_valid_sample_groups", "purpose": "Confirm available group columns and verify sample counts.", "status": "available", "args_hint": {"metadata_path": ""}, }, { "step": step_n + 1, "tool": "dataset_validate_contrast", "purpose": "Check that both groups have enough samples before DE.", "status": "available", "args_hint": { "group_column": contrast.get("design_factor", ""), "test_group": contrast.get("test_group", ""), "control_group": contrast.get("control_group", ""), }, }, { "step": step_n + 2, "tool": "decoupler_differential_expression", "purpose": "Gene-level differential expression.", "status": "available", "args_hint": { "method": de_method, "contrast": [ contrast.get("design_factor", "condition"), contrast.get("test_group", ""), contrast.get("control_group", ""), ], }, }, { "step": step_n + 3, "tool": "decoupler_tf_enrichment_collectri", "purpose": "TF activity enrichment from DE statistics.", "status": "available", "args_hint": {"organism": organism}, }, { "step": step_n + 4, "tool": "decoupler_pathway_enrichment_progeny", "purpose": "Pathway enrichment from DE statistics.", "status": "available", "args_hint": {"organism": organism}, }, { "step": step_n + 5, "tool": "decoupler_hallmark_enrichment", "purpose": "Hallmark gene set enrichment from DE statistics.", "status": "available", "args_hint": {"organism": organism}, }, ] required_inputs = [ "Expression file (h5ad, csv, or tsv) — samples as rows, genes as columns", "Sample metadata file with group labels", ] assumptions = [ f"Expression is pre-normalised (analysis_path='{analysis_path}').", "Gene columns are HGNC gene symbols.", "At least 3 samples per group.", ] if analysis_path == "B": assumptions.append( "Use method='ttest' or 'limma' — NOT 'deseq2' (data is pre-normalised)." ) warnings: list[str] = [] refusal_conditions: list[str] = [] elif intent == "score_samples": steps += [ { "step": step_n, "tool": "dataset_score_bulk_samples", "purpose": "Compute per-sample TF/pathway/hallmark activity scores.", "status": "available", "args_hint": {"resource": "progeny", "organism": organism, "method": "ulm"}, }, { "step": step_n + 1, "tool": "dataset_compare_activity_by_group", "purpose": "Optional: compare activity scores between groups.", "status": "available", "args_hint": { "group_column": contrast.get("design_factor", ""), "test_group": contrast.get("test_group", ""), "control_group": contrast.get("control_group", ""), }, }, ] required_inputs = [ "Expression file (h5ad, csv, or tsv) — samples as rows, genes as columns", ] assumptions = [ "Expression is pre-normalised.", "Gene columns are HGNC gene symbols.", ] warnings = [ "Set resource='progeny' for pathway scores, 'collectri' for TF, " "'hallmark' for gene sets.", ] refusal_conditions = [] elif intent == "survival": from src.workflows.survival import check_survival_data_available survival_available = ( check_survival_data_available(manifest)["available"] if manifest is not None else False ) survival_status = "available" if survival_available else "not_implemented" steps += [ { "step": step_n, "tool": "dataset_score_bulk_samples", "purpose": "Compute per-sample activity scores for survival modelling.", "status": "available", "args_hint": {"resource": "progeny", "organism": organism}, }, { "step": step_n + 1, "tool": "survival_analysis", "purpose": "Kaplan-Meier or Cox regression with activity as covariate.", "status": survival_status, "note": ( "Survival columns declared in manifest." if survival_available else "Survival data not in manifest. " "Obtain event+time columns from supplementary tables first." ), "args_hint": {}, }, ] required_inputs = [ "Expression file (h5ad, csv, or tsv)", "Survival metadata with numeric time and binary event (0/1) columns", ] assumptions = ["Expression is pre-normalised."] warnings = [ "Survival analysis is NOT YET IMPLEMENTED in this pipeline. " "Export activity scores (dataset_score_bulk_samples) and use " "lifelines or R survminer externally.", ] refusal_conditions = [ "Cannot run survival analysis until event and time columns are confirmed " "in the metadata.", ] elif intent == "correlate_continuous": steps += [ { "step": step_n, "tool": "dataset_score_bulk_samples", "purpose": "Compute per-sample activity scores as correlation input.", "status": "available", "args_hint": {"resource": "progeny", "organism": organism}, }, { "step": step_n + 1, "tool": "associate_activity_with_covariate", "purpose": "Correlate activity with a continuous covariate.", "status": "not_implemented", "note": ( "Not yet implemented. Export activity CSV and use " "scipy.stats.spearmanr or R cor.test externally." ), "args_hint": {}, }, ] required_inputs = [ "Expression file (h5ad, csv, or tsv)", "Metadata file with the continuous covariate column", ] assumptions = [ "Expression is pre-normalised.", "The covariate is a continuous numeric variable.", ] warnings = [ "Continuous covariate association is NOT YET IMPLEMENTED. " "Export activity scores and correlate externally.", ] refusal_conditions = [] else: # unknown steps = [ { "step": 1, "tool": "dataset_list_available", "purpose": "Discover what datasets are registered.", "status": "available", "args_hint": {}, }, { "step": 2, "tool": "dataset_describe", "purpose": "Understand the dataset before choosing an analysis.", "status": "available", "args_hint": {"dataset_id": dataset_id}, }, ] required_inputs = ["Clarification of the analysis goal."] assumptions = [] warnings = [ "Could not detect a specific analysis intent. " "Rephrase to include keywords like: 'compare', 'activity', " "'survival', or 'correlate'.", ] refusal_conditions = [ "Will not start analysis until the user's goal is clear.", ] return { "steps": steps, "required_inputs": required_inputs, "assumptions": assumptions, "warnings": warnings, "refusal_conditions": refusal_conditions, } # --------------------------------------------------------------------------- # Tool # --------------------------------------------------------------------------- def _col_semantics(col_def, col_name: str) -> dict: """Counting semantics for a metadata column def (MetadataColumnDef or dict).""" if hasattr(col_def, "missing_values"): return { "missing_set": set(col_def.missing_values), "allowed_set": set(col_def.allowed_values), "bio_ok": col_def.biological_grouping_allowed, "check_col": col_def.decoded_column or col_name, "interp_warn": col_def.interpretation_warning, "refusal_rules": list(col_def.refusal_rules or []), } return { "missing_set": set(col_def.get("missing_values") or []), "allowed_set": set(col_def.get("allowed_values") or []), "bio_ok": col_def.get("biological_grouping_allowed", True), "check_col": col_def.get("decoded_column") or col_name, "interp_warn": col_def.get("interpretation_warning", ""), "refusal_rules": list(col_def.get("refusal_rules") or []), } def _classify_metadata_values(raw_counts, missing_set: set, allowed_set: set): """Split a value->count mapping into (annotated, missing, unexpected) dicts.""" annotated: dict[str, int] = {} missing: dict[str, int] = {} unexpected: dict[str, int] = {} for val, cnt in raw_counts.items(): if val in missing_set or val in {"nan", "None", ""}: missing[val] = cnt elif allowed_set and val not in allowed_set: unexpected[val] = cnt else: annotated[val] = cnt return annotated, missing, unexpected def _flatten_single_column_summary( s: dict, col_def, manifest, column: str, adata_path: str, dataset_id: str ) -> dict: """Flattened top-level fields when a single column was requested.""" empty_meaning = ( col_def.empty_value_meaning if hasattr(col_def, "empty_value_meaning") else col_def.get("empty_value_meaning", "") ) interp_warn = ( col_def.interpretation_warning if hasattr(col_def, "interpretation_warning") else col_def.get("interpretation_warning", "") ) warnings: list[str] = [] if empty_meaning: warnings.append(empty_meaning.strip()) if interp_warn: warnings.append(interp_warn.strip()) # missing_counts: replace raw empty-string key with the human meaning missing_counts: dict[str, int] = {} for val, cnt in s["missing"].items(): label = empty_meaning.split(".")[0].strip() if empty_meaning and val == "" else val missing_counts[label or val] = cnt # Suggest crosstab when there are missing values. crosstab_next: str | None = None if s["total_missing"] > 0: group_cols = getattr(manifest, "group_columns", []) other_cols = [c for c in group_cols if c != column] crosstab_col = other_cols[0] if other_cols else "cell_line/tissue" crosstab_next = ( f"To understand what the {s['total_missing']} unannotated samples are, call: " f"dataset_crosstab_metadata_values(" f"adata_path='{adata_path}', " f"row_column='{column}', " f"col_column='{crosstab_col}', " f"dataset_id='{dataset_id}')" ) return { "column": s["column"], "n_total": s["total"], "n_annotated": s["total_annotated"], "n_missing": s["total_missing"], "annotated_counts": s["annotated"], "missing_counts": missing_counts, "warnings": warnings, "prohibited_inferences": s.get("prohibited_inferences", []), "next_step_if_missing": crosstab_next, }