298 lines
10 KiB
Python
Executable File
298 lines
10 KiB
Python
Executable File
"""Shared STRING/PPI policy helpers.
|
|
|
|
This module intentionally stays small: it centralizes species aliases and
|
|
stage-local mapping thresholds without owning PPI execution or Ortholog.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional
|
|
|
|
|
|
# Verified with STRING v12 API get_string_ids on 2026-07-07:
|
|
# species=4530 resolves rice query IDs to ncbiTaxonId 39947 (Oryza sativa Japonica).
|
|
STRING_SPECIES_MAP: Dict[str, str] = {
|
|
"ath": "3702",
|
|
"soe": "3562",
|
|
"osa": "39947",
|
|
"zma": "4577",
|
|
"gma": "3847",
|
|
"ptr": "3694",
|
|
"vvi": "29760",
|
|
"sly": "4081",
|
|
"mtr": "3880",
|
|
"stu": "4113",
|
|
"bdi": "29655",
|
|
"sbi": "4558",
|
|
}
|
|
|
|
|
|
STRING_SPECIES_ALIASES: Dict[str, str] = {
|
|
**{code: code for code in STRING_SPECIES_MAP},
|
|
"at": "ath",
|
|
"ath": "ath",
|
|
"arabidopsis": "ath",
|
|
"arabidopsis_thaliana": "ath",
|
|
"arabidopsis thaliana": "ath",
|
|
"soe": "soe",
|
|
"spinacia": "soe",
|
|
"spinacia_oleracea": "soe",
|
|
"spinacia oleracea": "soe",
|
|
"spinach": "soe",
|
|
"os": "osa",
|
|
"osa": "osa",
|
|
"rice": "osa",
|
|
"oryza_sativa": "osa",
|
|
"oryza sativa": "osa",
|
|
"zma": "zma",
|
|
"zm": "zma",
|
|
"maize": "zma",
|
|
"zea_mays": "zma",
|
|
"zea mays": "zma",
|
|
"gma": "gma",
|
|
"gm": "gma",
|
|
"soybean": "gma",
|
|
"glycine_max": "gma",
|
|
"glycine max": "gma",
|
|
"ptr": "ptr",
|
|
"pt": "ptr",
|
|
"populus": "ptr",
|
|
"populus_trichocarpa": "ptr",
|
|
"populus trichocarpa": "ptr",
|
|
"vvi": "vvi",
|
|
"vv": "vvi",
|
|
"grape": "vvi",
|
|
"vitis_vinifera": "vvi",
|
|
"vitis vinifera": "vvi",
|
|
"sly": "sly",
|
|
"tomato": "sly",
|
|
"solanum_lycopersicum": "sly",
|
|
"solanum lycopersicum": "sly",
|
|
}
|
|
|
|
STRING_REFERENCE_RECOMMENDATIONS: Dict[str, List[Dict[str, str]]] = {
|
|
# Fagopyrum tataricum is not present in STRING v12. Spinacia oleracea is
|
|
# STRING-supported and in the same order (Caryophyllales), making it a
|
|
# closer reference than Arabidopsis for ortholog projection.
|
|
"ft": [
|
|
{
|
|
"species_code": "soe",
|
|
"species_name": "Spinacia oleracea",
|
|
"taxon_id": "3562",
|
|
"relationship": "same_order_caryophyllales",
|
|
"source": "STRING v12 species list",
|
|
"confidence": "near_reference",
|
|
}
|
|
],
|
|
"fagopyrum tataricum": [
|
|
{
|
|
"species_code": "soe",
|
|
"species_name": "Spinacia oleracea",
|
|
"taxon_id": "3562",
|
|
"relationship": "same_order_caryophyllales",
|
|
"source": "STRING v12 species list",
|
|
"confidence": "near_reference",
|
|
}
|
|
],
|
|
"fagopyrum_tataricum": [
|
|
{
|
|
"species_code": "soe",
|
|
"species_name": "Spinacia oleracea",
|
|
"taxon_id": "3562",
|
|
"relationship": "same_order_caryophyllales",
|
|
"source": "STRING v12 species list",
|
|
"confidence": "near_reference",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def species_to_string_code(species: str, fallback: str = "ath") -> str:
|
|
token = str(species or "").strip().lower()
|
|
if not token:
|
|
return fallback
|
|
return STRING_SPECIES_ALIASES.get(token, fallback)
|
|
|
|
|
|
def recommended_string_reference_species(target_species: str) -> List[Dict[str, str]]:
|
|
token = str(target_species or "").strip().lower()
|
|
if not token:
|
|
return []
|
|
return [dict(item) for item in STRING_REFERENCE_RECOMMENDATIONS.get(token, [])]
|
|
|
|
|
|
def string_taxon_id(species_code: str, fallback: str = "3702") -> str:
|
|
code = species_to_string_code(species_code, "")
|
|
return STRING_SPECIES_MAP.get(code, fallback)
|
|
|
|
|
|
def unique_preserve_order(values: List[str]) -> List[str]:
|
|
seen = set()
|
|
unique = []
|
|
for value in values:
|
|
if value and value not in seen:
|
|
unique.append(value)
|
|
seen.add(value)
|
|
return unique
|
|
|
|
|
|
def normalize_gene_ids_for_string(gene_ids: List[str], species: str) -> List[str]:
|
|
"""Convert local transcript IDs to stage-local STRING query IDs."""
|
|
species_code = species_to_string_code(species, species)
|
|
normalized = []
|
|
for gene_id in gene_ids:
|
|
value = str(gene_id or "").strip()
|
|
if species_code == "osa":
|
|
match = re.match(r"^(Os\d{2})t(\d+)(?:-\d+)?$", value, flags=re.IGNORECASE)
|
|
if match:
|
|
value = f"{match.group(1)}g{match.group(2)}"
|
|
normalized.append(value)
|
|
return unique_preserve_order(normalized)
|
|
|
|
|
|
def reference_protein_candidate_paths(species: str, roots: Optional[List[Path]] = None) -> List[Path]:
|
|
"""Return conservative local reference proteome candidates for PPI fallback."""
|
|
roots = roots or [Path("/data/ref_genome"), Path("/data")]
|
|
raw = str(species or "").strip()
|
|
code = species_to_string_code(raw, raw.lower())
|
|
tokens = unique_preserve_order([raw, raw.lower(), code])
|
|
suffixes = [
|
|
"{token}.pep.fasta",
|
|
"{token}.pep.fa",
|
|
"{token}.protein.fasta",
|
|
"{token}.protein.fa",
|
|
"{token}.faa",
|
|
]
|
|
candidates: List[Path] = []
|
|
for root in roots:
|
|
for token in tokens:
|
|
if not token:
|
|
continue
|
|
for suffix in suffixes:
|
|
name = suffix.format(token=token)
|
|
candidates.append(root / token / name)
|
|
candidates.append(root / name)
|
|
return unique_preserve_order(candidates)
|
|
|
|
|
|
def evaluate_string_mapping_quality(input_count: int, mapped_count: int) -> Dict[str, object]:
|
|
input_count = max(int(input_count or 0), 0)
|
|
mapped_count = max(int(mapped_count or 0), 0)
|
|
mapping_rate = (mapped_count / input_count) if input_count else 0.0
|
|
required_count = min(3, input_count)
|
|
ok = input_count > 0 and mapped_count >= required_count and mapping_rate >= 0.5
|
|
if input_count <= 2:
|
|
ok = input_count > 0 and mapped_count == input_count
|
|
return {
|
|
"status": "success" if ok else "failed",
|
|
"issue_code": "" if ok else "ppi_string_id_mapping_low",
|
|
"input_count": input_count,
|
|
"mapped_count": mapped_count,
|
|
"mapping_rate": round(mapping_rate, 6),
|
|
"required_mapped_count": required_count,
|
|
}
|
|
|
|
|
|
def build_ppi_species_diagnostic(
|
|
*,
|
|
requested_ppi_species: str,
|
|
target_species: str,
|
|
target_string_supported: bool,
|
|
reference_species: str = "",
|
|
reference_data_source: str = "",
|
|
ortholog_mapping_used: bool = False,
|
|
fallback_level: str = "",
|
|
fallback_reason: str = "",
|
|
selected_string_species: str = "",
|
|
selected_taxon_id: str = "",
|
|
prefix: str = "",
|
|
artifacts: Optional[Dict[str, object]] = None,
|
|
) -> Dict[str, object]:
|
|
"""Build a failed PPI manifest fragment for missing species/mapping inputs.
|
|
|
|
PPI execution does not run Ortholog or download references. This diagnostic
|
|
tells the planner/user what must happen before PPI can run.
|
|
"""
|
|
reference_available = bool(reference_data_source)
|
|
issue_code = "ppi_requires_ortholog_mapping" if reference_available else "ppi_reference_data_missing"
|
|
if ortholog_mapping_used:
|
|
issue_code = "ppi_invalid_ortholog_mapping"
|
|
recommendation = (
|
|
"Run Ortholog as a separate stage using the available reference protein FASTA, then rerun PPI."
|
|
if reference_available
|
|
else "Provide or select a near/reference species protein FASTA, then run Ortholog before PPI."
|
|
)
|
|
if reference_available:
|
|
recommended_actions = [
|
|
{
|
|
"id": "run_ortholog",
|
|
"label": "Run Ortholog before PPI",
|
|
"requires_confirmation": False,
|
|
"risk": "required_mapping_step",
|
|
},
|
|
{
|
|
"id": "change_reference_species",
|
|
"label": "Select a different reference species with protein FASTA",
|
|
"requires_confirmation": True,
|
|
"risk": "reference_choice_changes_network",
|
|
},
|
|
]
|
|
else:
|
|
recommended_actions = [
|
|
{
|
|
"id": "upload_reference_protein",
|
|
"label": "Upload a near/reference species protein FASTA",
|
|
"requires_confirmation": False,
|
|
"risk": "requires_user_data",
|
|
},
|
|
{
|
|
"id": "prepare_reference_download",
|
|
"label": "Prepare a provenance-recorded reference protein download",
|
|
"requires_confirmation": True,
|
|
"risk": "network_and_release_dependent",
|
|
},
|
|
{
|
|
"id": "explicit_arabidopsis_last_resort",
|
|
"label": "Use Arabidopsis as an explicit last-resort model reference",
|
|
"requires_confirmation": True,
|
|
"risk": "low_confidence_model_reference",
|
|
},
|
|
]
|
|
return {
|
|
"manifest_version": "1.0",
|
|
"stage": "ppi",
|
|
"status": "failed",
|
|
"prefix": prefix,
|
|
"parameters": {
|
|
"ppi_species_resolution": {
|
|
"requested_ppi_species": requested_ppi_species or "auto",
|
|
"target_species": target_species,
|
|
"target_string_supported": bool(target_string_supported),
|
|
"selected_string_species": selected_string_species,
|
|
"selected_taxon_id": selected_taxon_id,
|
|
"reference_species": reference_species,
|
|
"reference_data_source": reference_data_source,
|
|
"ortholog_mapping_used": bool(ortholog_mapping_used),
|
|
"ortholog_mapping_source": "",
|
|
"fallback_level": fallback_level,
|
|
"fallback_reason": fallback_reason,
|
|
"decision": "failed_requires_ortholog",
|
|
}
|
|
},
|
|
"metrics": {
|
|
"id_resolution": {},
|
|
"string_image_status": "skipped",
|
|
},
|
|
"diagnostics": {
|
|
"issue_code": issue_code,
|
|
"severity": "blocking",
|
|
"layer": "input_contract",
|
|
"required_next_stage": "ortholog",
|
|
"recommendation": recommendation,
|
|
"recommended_actions": recommended_actions,
|
|
},
|
|
"artifacts": artifacts or {},
|
|
}
|