1380 lines
51 KiB
Python
Executable File
1380 lines
51 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
PPI Skill - Protein-Protein Interaction Analysis
|
|
蛋白互作网络分析技能 (基于STRING数据库)
|
|
|
|
Usage:
|
|
python run.py --gene-ids genes.txt --species ath --prefix Ft.SPL --output ./ppi
|
|
python run.py --config config.yaml --output ./ppi
|
|
python run.py --demo --output ./ppi
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Dict, List, Optional
|
|
|
|
try:
|
|
import requests
|
|
HAS_REQUESTS = True
|
|
except ImportError:
|
|
HAS_REQUESTS = False
|
|
|
|
import pandas as pd
|
|
import yaml
|
|
|
|
from gfa_core.id_resolution import IdResolutionError, ReferenceAliasIndex
|
|
from gfa_core.ppi_string_policy import (
|
|
STRING_SPECIES_MAP,
|
|
normalize_gene_ids_for_string,
|
|
string_taxon_id,
|
|
unique_preserve_order,
|
|
)
|
|
from gfa_core.string_rate_gate import (
|
|
StringRateGate,
|
|
is_authoritative_string_https_url,
|
|
string_https_request_identity,
|
|
)
|
|
from gfa_core.string_species_registry import StringSnapshotError, load_string_snapshot
|
|
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StringApiError(RuntimeError):
|
|
"""Typed failure from a STRING transport, HTTP, JSON, or protocol boundary."""
|
|
|
|
def __init__(
|
|
self,
|
|
message: str,
|
|
*,
|
|
kind: str,
|
|
status_code: int | None = None,
|
|
endpoint: str = "",
|
|
) -> None:
|
|
super().__init__(message)
|
|
self.kind = kind
|
|
self.status_code = status_code
|
|
self.endpoint = endpoint
|
|
|
|
|
|
class StringAPI:
|
|
"""STRING API client with stable-version discovery and typed failures."""
|
|
|
|
BASE_URL = "https://string-db.org/api"
|
|
CALLER_IDENTITY = "PlantAI"
|
|
SPECIES_MAP = STRING_SPECIES_MAP
|
|
|
|
def __init__(
|
|
self,
|
|
species: str = "ath",
|
|
*,
|
|
http_client: Any = None,
|
|
session: Any = None,
|
|
stable_api_base: str | None = None,
|
|
rate_gate: StringRateGate | Any | None = None,
|
|
rate_gate_path: Path | str | None = None,
|
|
min_interval: float = 1.0,
|
|
clock: Callable[[], float] | None = None,
|
|
sleeper: Callable[[float], None] | None = None,
|
|
):
|
|
self.species = species
|
|
self.species_id = self.resolve_species_id(species)
|
|
self._http = http_client or session or (requests if HAS_REQUESTS else None)
|
|
self.stable_api_base = self._normalize_api_base(stable_api_base or "")
|
|
gate_options: Dict[str, Any] = {
|
|
"path": rate_gate_path,
|
|
"min_interval": min_interval,
|
|
}
|
|
if clock is not None:
|
|
gate_options["clock"] = clock
|
|
if sleeper is not None:
|
|
gate_options["sleeper"] = sleeper
|
|
self.rate_gate = rate_gate or StringRateGate(**gate_options)
|
|
|
|
@staticmethod
|
|
def resolve_species_id(species: str) -> str:
|
|
raw = str(species or "").strip()
|
|
if raw.isdigit():
|
|
snapshot_root = Path(__file__).resolve().parents[2] / "data" / "string" / "v12.0"
|
|
try:
|
|
snapshot = load_string_snapshot(snapshot_root)
|
|
except StringSnapshotError as exc:
|
|
raise ValueError(f"STRING snapshot registry unavailable: {exc}") from exc
|
|
if raw not in snapshot.registry:
|
|
raise ValueError(f"STRING TaxID {raw} is not in the fixed STRING snapshot registry")
|
|
return raw
|
|
species_id = string_taxon_id(species, "")
|
|
if not species_id:
|
|
raise ValueError(f"Unsupported or empty STRING species: {species!r}")
|
|
return species_id
|
|
|
|
@staticmethod
|
|
def _normalize_api_base(address: str) -> str:
|
|
value = str(address or "").strip()
|
|
if not value:
|
|
return ""
|
|
try:
|
|
parsed = urllib.parse.urlsplit(value)
|
|
except ValueError as exc:
|
|
raise StringApiError(
|
|
"Malformed STRING stable address", kind="protocol", endpoint="version"
|
|
) from exc
|
|
hostname = (parsed.hostname or "").lower()
|
|
if (
|
|
not is_authoritative_string_https_url(value)
|
|
or parsed.query
|
|
or parsed.fragment
|
|
or parsed.path not in {"", "/", "/api", "/api/"}
|
|
):
|
|
raise StringApiError(
|
|
"Untrusted STRING stable address", kind="protocol", endpoint="version"
|
|
)
|
|
return f"https://{hostname}/api"
|
|
|
|
def _request_json(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
allow_404: bool = False,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
if self._http is None:
|
|
raise StringApiError(
|
|
"requests module not available", kind="transport", endpoint=url
|
|
)
|
|
self.rate_gate.wait()
|
|
try:
|
|
request_method = getattr(self._http, method.lower())
|
|
response = request_method(
|
|
url, timeout=30, allow_redirects=False, **kwargs
|
|
)
|
|
except Exception as exc:
|
|
raise StringApiError(
|
|
f"STRING transport failure: {exc}", kind="transport", endpoint=url
|
|
) from exc
|
|
|
|
response_url = getattr(response, "url", None)
|
|
if string_https_request_identity(response_url) != string_https_request_identity(
|
|
url
|
|
):
|
|
raise StringApiError(
|
|
f"STRING response URL does not match request: {response_url!r}",
|
|
kind="protocol",
|
|
endpoint=url,
|
|
)
|
|
raw_status = getattr(response, "status_code", None)
|
|
if isinstance(raw_status, bool) or not isinstance(raw_status, int):
|
|
raise StringApiError(
|
|
f"Malformed STRING HTTP status: {raw_status!r}",
|
|
kind="protocol",
|
|
endpoint=url,
|
|
)
|
|
status_code = raw_status
|
|
if allow_404 and status_code == 404:
|
|
return []
|
|
if status_code < 200 or status_code >= 300:
|
|
raise StringApiError(
|
|
f"STRING HTTP {status_code}",
|
|
kind="http",
|
|
status_code=status_code,
|
|
endpoint=url,
|
|
)
|
|
try:
|
|
return response.json()
|
|
except Exception as exc:
|
|
raise StringApiError(
|
|
f"STRING returned invalid JSON: {exc}", kind="json", endpoint=url
|
|
) from exc
|
|
|
|
def get_version(self) -> Dict[str, Any]:
|
|
"""Discover and retain the stable API address for this STRING version."""
|
|
payload = self._request_json(
|
|
"GET",
|
|
f"{self.BASE_URL}/json/version",
|
|
params={"caller_identity": self.CALLER_IDENTITY},
|
|
)
|
|
if isinstance(payload, list) and len(payload) == 1:
|
|
version = payload[0]
|
|
elif isinstance(payload, dict):
|
|
version = payload
|
|
else:
|
|
raise StringApiError(
|
|
"STRING version response must be one object",
|
|
kind="protocol",
|
|
endpoint="version",
|
|
)
|
|
if not isinstance(version, dict):
|
|
raise StringApiError(
|
|
"STRING version response contains a non-object",
|
|
kind="protocol",
|
|
endpoint="version",
|
|
)
|
|
stable_address = version.get("string_stable_address") or version.get(
|
|
"stable_address"
|
|
)
|
|
if not isinstance(stable_address, str) or not stable_address.strip():
|
|
raise StringApiError(
|
|
"STRING version response has no stable address",
|
|
kind="protocol",
|
|
endpoint="version",
|
|
)
|
|
self.stable_api_base = self._normalize_api_base(stable_address)
|
|
return dict(version)
|
|
|
|
def _stable_base(self) -> str:
|
|
if not self.stable_api_base:
|
|
self.get_version()
|
|
return self.stable_api_base
|
|
|
|
@classmethod
|
|
def validate_mapping_rows(
|
|
cls,
|
|
payload: Any,
|
|
identifiers: List[str],
|
|
species_id: str,
|
|
) -> List[Dict[str, Any]]:
|
|
if not isinstance(payload, list):
|
|
raise StringApiError(
|
|
"STRING mapping response must be a list", kind="protocol"
|
|
)
|
|
seen_indices = set()
|
|
validated = []
|
|
expected_taxon = int(species_id)
|
|
for row in payload:
|
|
if not isinstance(row, dict):
|
|
raise StringApiError(
|
|
"STRING mapping row must be an object", kind="protocol"
|
|
)
|
|
query_index = row.get("queryIndex")
|
|
if (
|
|
isinstance(query_index, bool)
|
|
or not isinstance(query_index, int)
|
|
or query_index < 0
|
|
or query_index >= len(identifiers)
|
|
or query_index in seen_indices
|
|
):
|
|
raise StringApiError(
|
|
f"Invalid STRING queryIndex: {query_index!r}", kind="protocol"
|
|
)
|
|
seen_indices.add(query_index)
|
|
taxon = row.get("ncbiTaxonId")
|
|
parsed_taxon = cls._validated_taxon(taxon)
|
|
if parsed_taxon != expected_taxon:
|
|
raise StringApiError(
|
|
f"STRING mapping taxon {parsed_taxon} != {expected_taxon}",
|
|
kind="protocol",
|
|
)
|
|
cls._validated_string_id(row.get("stringId"), parsed_taxon)
|
|
if (
|
|
not isinstance(row.get("queryItem"), str)
|
|
or row["queryItem"] != identifiers[query_index]
|
|
or not isinstance(row.get("stringId"), str)
|
|
or not row["stringId"].strip()
|
|
or not isinstance(row.get("preferredName"), str)
|
|
or not row["preferredName"].strip()
|
|
):
|
|
raise StringApiError("Malformed STRING mapping row", kind="protocol")
|
|
validated.append(dict(row))
|
|
return sorted(validated, key=lambda row: row["queryIndex"])
|
|
|
|
@staticmethod
|
|
def _validated_taxon(value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
raise StringApiError(f"Invalid STRING taxon: {value!r}", kind="protocol")
|
|
if isinstance(value, int):
|
|
return value
|
|
if isinstance(value, str) and value.isdigit():
|
|
return int(value)
|
|
raise StringApiError(f"Invalid STRING taxon: {value!r}", kind="protocol")
|
|
|
|
@staticmethod
|
|
def _validated_string_id(value: Any, taxon_id: int) -> str:
|
|
if not isinstance(value, str) or value != value.strip():
|
|
raise StringApiError(
|
|
f"Invalid STRING identifier: {value!r}", kind="protocol"
|
|
)
|
|
prefix, separator, entity = value.partition(".")
|
|
if (
|
|
separator != "."
|
|
or prefix != str(taxon_id)
|
|
or not entity
|
|
or any(character.isspace() for character in entity)
|
|
):
|
|
raise StringApiError(
|
|
f"Invalid STRING identifier for taxon {taxon_id}: {value!r}",
|
|
kind="protocol",
|
|
)
|
|
return value
|
|
|
|
def get_string_ids(self, identifiers: List[str]) -> List[Dict[str, Any]]:
|
|
"""Resolve candidate identifiers, attributing rows only by queryIndex."""
|
|
self._validate_identifiers(identifiers)
|
|
url = f"{self._stable_base()}/json/get_string_ids"
|
|
payload = self._request_json(
|
|
"POST",
|
|
url,
|
|
allow_404=True,
|
|
data={
|
|
"identifiers": "\r\n".join(identifiers),
|
|
"species": self.species_id,
|
|
"echo_query": 1,
|
|
"caller_identity": self.CALLER_IDENTITY,
|
|
},
|
|
)
|
|
return self.validate_mapping_rows(payload, identifiers, self.species_id)
|
|
|
|
@staticmethod
|
|
def _validate_identifiers(identifiers: List[str]) -> None:
|
|
if not isinstance(identifiers, list) or not identifiers:
|
|
raise ValueError("identifiers must be a non-empty list")
|
|
if any(not isinstance(item, str) or not item.strip() for item in identifiers):
|
|
raise ValueError("identifiers must contain non-empty strings")
|
|
|
|
def get_network(
|
|
self,
|
|
proteins: List[str],
|
|
required_score: int = 200,
|
|
network_type: str = "functional",
|
|
) -> List[Dict[str, Any]]:
|
|
"""Return a validated STRING network; an empty list is scientifically valid."""
|
|
self._validate_identifiers(proteins)
|
|
if (
|
|
isinstance(required_score, bool)
|
|
or not isinstance(required_score, int)
|
|
or not 0 <= required_score <= 1000
|
|
):
|
|
raise ValueError("required_score must be an integer from 0 to 1000")
|
|
if network_type != "functional":
|
|
raise ValueError("network_type is fixed to 'functional'")
|
|
selected_taxon = int(self.species_id)
|
|
for protein in proteins:
|
|
self._validated_string_id(protein, selected_taxon)
|
|
submitted_ids = set(proteins)
|
|
url = f"{self._stable_base()}/json/network"
|
|
payload = self._request_json(
|
|
"POST",
|
|
url,
|
|
data={
|
|
"identifiers": "\r\n".join(proteins),
|
|
"species": self.species_id,
|
|
"required_score": required_score,
|
|
"network_type": "functional",
|
|
"add_nodes": 0,
|
|
"caller_identity": self.CALLER_IDENTITY,
|
|
},
|
|
)
|
|
if not isinstance(payload, list):
|
|
raise StringApiError(
|
|
"STRING network response must be a list",
|
|
kind="protocol",
|
|
endpoint=url,
|
|
)
|
|
for row in payload:
|
|
if not isinstance(row, dict):
|
|
raise StringApiError(
|
|
"Malformed STRING network row", kind="protocol", endpoint=url
|
|
)
|
|
parsed_taxon = self._validated_taxon(row.get("ncbiTaxonId"))
|
|
preferred_a = row.get("preferredName_A")
|
|
preferred_b = row.get("preferredName_B")
|
|
score = row.get("score")
|
|
if (
|
|
parsed_taxon != int(self.species_id)
|
|
or not isinstance(preferred_a, str)
|
|
or not preferred_a.strip()
|
|
or not isinstance(preferred_b, str)
|
|
or not preferred_b.strip()
|
|
or isinstance(score, bool)
|
|
or not isinstance(score, (int, float))
|
|
or not math.isfinite(score)
|
|
or not 0 <= score <= 1
|
|
):
|
|
raise StringApiError(
|
|
"Malformed STRING network row", kind="protocol", endpoint=url
|
|
)
|
|
string_id_a = self._validated_string_id(
|
|
row.get("stringId_A"), parsed_taxon
|
|
)
|
|
string_id_b = self._validated_string_id(
|
|
row.get("stringId_B"), parsed_taxon
|
|
)
|
|
if (
|
|
string_id_a not in submitted_ids
|
|
or string_id_b not in submitted_ids
|
|
or score < required_score / 1000
|
|
):
|
|
raise StringApiError(
|
|
"STRING network row contradicts submitted identifiers or score threshold",
|
|
kind="protocol",
|
|
endpoint=url,
|
|
)
|
|
return payload
|
|
|
|
|
|
def generate_string_url(proteins: List[str], species: str = "3702", required_score: int = 200) -> str:
|
|
"""生成STRING网页链接
|
|
|
|
Args:
|
|
proteins: 蛋白ID列表
|
|
species: STRING物种ID (默认3702=拟南芥)
|
|
|
|
Returns:
|
|
STRING网络查看URL
|
|
"""
|
|
base_url = "https://string-db.org/cgi/network"
|
|
# 使用实际换行符，urllib.parse.urlencode会自动编码
|
|
identifiers = "\r\n".join(proteins)
|
|
params = {
|
|
"identifiers": identifiers,
|
|
"species": species,
|
|
"required_score": str(required_score),
|
|
"network_type": "functional"
|
|
}
|
|
return f"{base_url}?{urllib.parse.urlencode(params)}"
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PpiQueryOrigin:
|
|
canonical_target_id: str
|
|
ortholog_reference_id: str
|
|
raw_query_id: str
|
|
relationship_source: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PpiResolutionRow:
|
|
canonical_target_id: str
|
|
ortholog_reference_id: str
|
|
raw_query_id: str
|
|
candidate_gene_id: str
|
|
selected_query_id: str
|
|
string_id: str
|
|
string_preferred_name: str
|
|
mapping_status: str
|
|
method: str
|
|
evidence_source: str
|
|
relationship_source: str
|
|
species: str
|
|
taxon_id: str
|
|
reason: str
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class StringResolutionResult:
|
|
selected_string_ids: List[str]
|
|
rows: List[PpiResolutionRow]
|
|
metrics: Dict[str, Any]
|
|
status: str = "success"
|
|
issue_code: str = ""
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _ResolutionCandidate:
|
|
value: str
|
|
namespace: str
|
|
method: str
|
|
evidence_source: str
|
|
|
|
|
|
def parse_ortholog_mapping(mapping_file: Path) -> List[PpiQueryOrigin]:
|
|
"""Parse every canonical-to-reference relationship without rewriting IDs."""
|
|
mapping: List[PpiQueryOrigin] = []
|
|
if not mapping_file.exists():
|
|
return mapping
|
|
|
|
with open(mapping_file, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
parts = line.split("\t")
|
|
if len(parts) >= 2:
|
|
source_id = parts[0].strip()
|
|
target_id = parts[1].strip()
|
|
if source_id and target_id:
|
|
mapping.append(
|
|
PpiQueryOrigin(
|
|
canonical_target_id=source_id,
|
|
ortholog_reference_id=target_id,
|
|
raw_query_id=target_id,
|
|
relationship_source=str(mapping_file),
|
|
)
|
|
)
|
|
|
|
return mapping
|
|
|
|
|
|
class PpiSkill:
|
|
"""PPI分析技能主类"""
|
|
|
|
def __init__(
|
|
self,
|
|
config_path: Optional[str] = None,
|
|
*,
|
|
string_api_factory: Callable[[str], Any] | None = None,
|
|
):
|
|
self.config = self._load_config(config_path)
|
|
self.checksums = {}
|
|
self.start_time = None
|
|
self.end_time = None
|
|
self.ortholog_mapping: List[PpiQueryOrigin] | None = None
|
|
self.string_api_factory = string_api_factory or StringAPI
|
|
|
|
def _load_config(self, config_path: Optional[str]) -> Dict:
|
|
"""加载配置文件"""
|
|
if config_path and os.path.exists(config_path):
|
|
with open(config_path) as f:
|
|
return yaml.safe_load(f)
|
|
default_config = Path(__file__).parent / "config.yaml"
|
|
with open(default_config) as f:
|
|
return yaml.safe_load(f)
|
|
|
|
def _compute_checksum(self, filepath: Path) -> str:
|
|
"""计算文件SHA256校验"""
|
|
sha256 = hashlib.sha256()
|
|
with open(filepath, "rb") as f:
|
|
for chunk in iter(lambda: f.read(8192), b""):
|
|
sha256.update(chunk)
|
|
return sha256.hexdigest()
|
|
|
|
def _record_file(self, filepath: Path, file_type: str = "output") -> Dict:
|
|
"""记录文件信息"""
|
|
if filepath.exists():
|
|
return {
|
|
"path": str(filepath),
|
|
"sha256": self._compute_checksum(filepath),
|
|
"size": filepath.stat().st_size,
|
|
"type": file_type
|
|
}
|
|
return {}
|
|
|
|
def load_gene_ids(self, gene_file: Path) -> List[str]:
|
|
"""加载基因ID列表"""
|
|
gene_ids = []
|
|
with open(gene_file, 'r') as f:
|
|
for line in f:
|
|
gene_id = line.strip()
|
|
if gene_id:
|
|
gene_ids.append(gene_id)
|
|
return gene_ids
|
|
|
|
def load_ortholog_mapping(self, mapping_file: Path) -> List[PpiQueryOrigin]:
|
|
"""加载同源基因映射"""
|
|
self.ortholog_mapping = parse_ortholog_mapping(mapping_file)
|
|
logger.info(f"Loaded {len(self.ortholog_mapping)} ortholog mappings from {mapping_file}")
|
|
return self.ortholog_mapping
|
|
|
|
@staticmethod
|
|
def _query_candidates(
|
|
raw_query_id: str,
|
|
normalization_species_label: str,
|
|
alias_index: ReferenceAliasIndex | None,
|
|
) -> List[_ResolutionCandidate]:
|
|
raw = str(raw_query_id or "").strip()
|
|
candidates: List[_ResolutionCandidate] = []
|
|
|
|
def add(value: str, namespace: str, method: str, evidence_source: str) -> None:
|
|
token = str(value or "").strip()
|
|
if token and all(candidate.value != token for candidate in candidates):
|
|
candidates.append(
|
|
_ResolutionCandidate(token, namespace, method, evidence_source)
|
|
)
|
|
|
|
add(raw, "raw", "raw_query", "raw_query")
|
|
normalized = normalize_gene_ids_for_string([raw], normalization_species_label)
|
|
if normalized:
|
|
add(
|
|
normalized[0],
|
|
"normalized",
|
|
f"species_rule:{normalization_species_label}",
|
|
"ppi_string_policy",
|
|
)
|
|
if "." in raw and raw.rsplit(".", 1)[1].isdigit():
|
|
add(
|
|
raw.rsplit(".", 1)[0],
|
|
"versionless",
|
|
"versionless_lookup_candidate",
|
|
"raw_query_id",
|
|
)
|
|
if alias_index is not None:
|
|
for candidate in alias_index.candidates(raw):
|
|
add(
|
|
candidate.value,
|
|
candidate.namespace,
|
|
candidate.method,
|
|
candidate.evidence_source,
|
|
)
|
|
return candidates
|
|
|
|
def resolve_query_ids(
|
|
self,
|
|
origins: List[PpiQueryOrigin],
|
|
species: str,
|
|
id_map_file: Path | None,
|
|
*,
|
|
api: Any = None,
|
|
canonical_target_ids: List[str] | None = None,
|
|
normalization_species_label: str = "",
|
|
) -> StringResolutionResult:
|
|
"""Resolve relationship queries without collapsing their provenance rows."""
|
|
species_id = StringAPI.resolve_species_id(species)
|
|
if not isinstance(origins, list) or any(
|
|
not isinstance(item, PpiQueryOrigin) for item in origins
|
|
):
|
|
raise ValueError("origins must be a list of PpiQueryOrigin")
|
|
alias_index = (
|
|
ReferenceAliasIndex.from_tsv(
|
|
Path(id_map_file),
|
|
str(normalization_species_label or species).strip(),
|
|
)
|
|
if id_map_file is not None
|
|
else None
|
|
)
|
|
unique_queries = unique_preserve_order(
|
|
[origin.raw_query_id for origin in origins if origin.raw_query_id]
|
|
)
|
|
candidates_by_query: Dict[str, List[_ResolutionCandidate]] = {}
|
|
query_errors: Dict[str, IdResolutionError] = {}
|
|
normalization_label = str(normalization_species_label or species).strip()
|
|
for raw_query in unique_queries:
|
|
try:
|
|
candidates_by_query[raw_query] = self._query_candidates(
|
|
raw_query, normalization_label, alias_index
|
|
)
|
|
except IdResolutionError as exc:
|
|
candidates_by_query[raw_query] = []
|
|
query_errors[raw_query] = exc
|
|
|
|
probes = unique_preserve_order(
|
|
[
|
|
candidate.value
|
|
for raw_query in unique_queries
|
|
for candidate in candidates_by_query[raw_query]
|
|
]
|
|
)
|
|
string_api = api or self.string_api_factory(species)
|
|
mapping_rows = string_api.get_string_ids(probes) if probes else []
|
|
validated_rows = StringAPI.validate_mapping_rows(
|
|
mapping_rows, probes, species_id
|
|
)
|
|
mapped_by_probe = {
|
|
probes[row["queryIndex"]]: row for row in validated_rows
|
|
}
|
|
|
|
query_results: Dict[str, Dict[str, Any]] = {}
|
|
for raw_query in unique_queries:
|
|
candidates = candidates_by_query[raw_query]
|
|
successful = [
|
|
(candidate, mapped_by_probe[candidate.value])
|
|
for candidate in candidates
|
|
if candidate.value in mapped_by_probe
|
|
]
|
|
string_ids = unique_preserve_order(
|
|
[row["stringId"] for _candidate, row in successful]
|
|
)
|
|
if raw_query in query_errors:
|
|
query_results[raw_query] = {
|
|
"status": "ambiguous",
|
|
"reason": query_errors[raw_query].code,
|
|
"string_id": "",
|
|
"preferred_name": "",
|
|
"selected_candidate": None,
|
|
"raw_resolved": False,
|
|
"geneid_fallback": False,
|
|
}
|
|
elif not successful:
|
|
query_results[raw_query] = {
|
|
"status": "unresolved",
|
|
"reason": "no_string_mapping",
|
|
"string_id": "",
|
|
"preferred_name": "",
|
|
"selected_candidate": None,
|
|
"raw_resolved": False,
|
|
"geneid_fallback": False,
|
|
}
|
|
elif len(string_ids) > 1:
|
|
query_results[raw_query] = {
|
|
"status": "ambiguous",
|
|
"reason": "candidate_string_ids_conflict",
|
|
"string_id": "",
|
|
"preferred_name": "",
|
|
"selected_candidate": None,
|
|
"raw_resolved": False,
|
|
"geneid_fallback": False,
|
|
}
|
|
else:
|
|
selected_candidate, selected_row = successful[0]
|
|
raw_resolved = any(
|
|
candidate.namespace in {"raw", "normalized", "versionless"}
|
|
for candidate, _row in successful
|
|
)
|
|
geneid_resolved = any(
|
|
candidate.namespace == "ncbi_gene_id"
|
|
for candidate, _row in successful
|
|
)
|
|
query_results[raw_query] = {
|
|
"status": "resolved",
|
|
"reason": "candidates_converged",
|
|
"string_id": string_ids[0],
|
|
"preferred_name": str(selected_row.get("preferredName") or ""),
|
|
"selected_candidate": selected_candidate,
|
|
"raw_resolved": raw_resolved,
|
|
"geneid_fallback": geneid_resolved and not raw_resolved,
|
|
}
|
|
|
|
rows = [
|
|
self._resolution_row(origin, query_results[origin.raw_query_id], species)
|
|
for origin in origins
|
|
]
|
|
rows = self._exclude_conflicting_orthologs(rows)
|
|
selected_string_ids = unique_preserve_order(
|
|
[row.string_id for row in rows if row.mapping_status == "resolved"]
|
|
)
|
|
metrics = self._resolution_metrics(
|
|
origins,
|
|
rows,
|
|
unique_queries,
|
|
query_results,
|
|
selected_string_ids,
|
|
canonical_target_ids=canonical_target_ids,
|
|
)
|
|
if selected_string_ids:
|
|
return StringResolutionResult(selected_string_ids, rows, metrics)
|
|
return StringResolutionResult(
|
|
selected_string_ids,
|
|
rows,
|
|
metrics,
|
|
status="id_resolution_failure",
|
|
issue_code="ppi_string_id_resolution_failed",
|
|
)
|
|
|
|
@staticmethod
|
|
def _resolution_row(
|
|
origin: PpiQueryOrigin,
|
|
query_result: Dict[str, Any],
|
|
species: str,
|
|
) -> PpiResolutionRow:
|
|
selected = query_result.get("selected_candidate")
|
|
return PpiResolutionRow(
|
|
canonical_target_id=origin.canonical_target_id,
|
|
ortholog_reference_id=origin.ortholog_reference_id,
|
|
raw_query_id=origin.raw_query_id,
|
|
candidate_gene_id=(
|
|
selected.value
|
|
if selected is not None
|
|
and selected.namespace in {"ncbi_gene_id", "gene_symbol"}
|
|
else ""
|
|
),
|
|
selected_query_id=selected.value if selected is not None else "",
|
|
string_id=query_result["string_id"],
|
|
string_preferred_name=query_result["preferred_name"],
|
|
mapping_status=query_result["status"],
|
|
method=selected.method if selected is not None else "",
|
|
evidence_source=(
|
|
(
|
|
"ortholog_relationship"
|
|
if origin.ortholog_reference_id
|
|
else "canonical_member_input"
|
|
)
|
|
if selected is not None and selected.namespace == "raw"
|
|
else selected.evidence_source
|
|
if selected is not None
|
|
else origin.relationship_source
|
|
),
|
|
relationship_source=origin.relationship_source,
|
|
species=species,
|
|
taxon_id=StringAPI.resolve_species_id(species),
|
|
reason=query_result["reason"],
|
|
)
|
|
|
|
@staticmethod
|
|
def _exclude_conflicting_orthologs(
|
|
rows: List[PpiResolutionRow],
|
|
) -> List[PpiResolutionRow]:
|
|
rows_by_canonical: Dict[str, List[PpiResolutionRow]] = {}
|
|
for row in rows:
|
|
rows_by_canonical.setdefault(row.canonical_target_id, []).append(row)
|
|
conflicting = set()
|
|
for canonical, canonical_rows in rows_by_canonical.items():
|
|
references = {row.raw_query_id for row in canonical_rows}
|
|
if len(references) < 2:
|
|
continue
|
|
string_ids = {
|
|
row.string_id
|
|
for row in canonical_rows
|
|
if row.mapping_status == "resolved" and row.string_id
|
|
}
|
|
if (
|
|
len(string_ids) != 1
|
|
or any(row.mapping_status != "resolved" for row in canonical_rows)
|
|
):
|
|
conflicting.add(canonical)
|
|
return [
|
|
PpiResolutionRow(
|
|
**{
|
|
**{
|
|
field: getattr(row, field)
|
|
for field in row.__dataclass_fields__
|
|
},
|
|
"candidate_gene_id": "",
|
|
"selected_query_id": "",
|
|
"string_id": "",
|
|
"string_preferred_name": "",
|
|
"mapping_status": "ambiguous_ortholog",
|
|
"method": "",
|
|
"reason": "ortholog_references_do_not_converge",
|
|
}
|
|
)
|
|
if row.canonical_target_id in conflicting
|
|
else row
|
|
for row in rows
|
|
]
|
|
|
|
@staticmethod
|
|
def _resolution_metrics(
|
|
origins: List[PpiQueryOrigin],
|
|
rows: List[PpiResolutionRow],
|
|
unique_queries: List[str],
|
|
query_results: Dict[str, Dict[str, Any]],
|
|
selected_string_ids: List[str],
|
|
*,
|
|
canonical_target_ids: List[str] | None = None,
|
|
) -> Dict[str, Any]:
|
|
input_members = unique_preserve_order(
|
|
list(canonical_target_ids or [])
|
|
+ [origin.canonical_target_id for origin in origins]
|
|
)
|
|
resolved_queries = sum(
|
|
result["status"] == "resolved" for result in query_results.values()
|
|
)
|
|
states_by_member: Dict[str, set[str]] = {}
|
|
for row in rows:
|
|
states_by_member.setdefault(row.canonical_target_id, set()).add(
|
|
row.mapping_status
|
|
)
|
|
resolved_members = sum(states == {"resolved"} for states in states_by_member.values())
|
|
ambiguous_members = sum(
|
|
bool(states.intersection({"ambiguous", "ambiguous_ortholog"}))
|
|
for states in states_by_member.values()
|
|
)
|
|
unresolved_members = len(input_members) - resolved_members - ambiguous_members
|
|
input_count = len(input_members)
|
|
query_count = len(unique_queries)
|
|
probe_status = (
|
|
"complete"
|
|
if query_count and resolved_queries == query_count
|
|
else "partial"
|
|
if resolved_queries
|
|
else "failed"
|
|
)
|
|
return {
|
|
"input_member_count": input_count,
|
|
"relationship_count": len(origins),
|
|
"unique_reference_query_count": query_count,
|
|
"raw_resolved_query_count": sum(
|
|
bool(result["raw_resolved"]) for result in query_results.values()
|
|
),
|
|
"geneid_fallback_resolved_query_count": sum(
|
|
bool(result["geneid_fallback"]) for result in query_results.values()
|
|
),
|
|
"resolved_reference_query_count": resolved_queries,
|
|
"selected_string_query_count": len(selected_string_ids),
|
|
"resolved_member_count": resolved_members,
|
|
"unresolved_member_count": unresolved_members,
|
|
"ambiguous_member_count": ambiguous_members,
|
|
"member_mapping_rate": round(resolved_members / input_count, 6)
|
|
if input_count
|
|
else 0,
|
|
"query_mapping_rate": round(resolved_queries / query_count, 6)
|
|
if query_count
|
|
else 0,
|
|
"probe_status": probe_status,
|
|
}
|
|
|
|
def write_string_resolution_report(
|
|
self,
|
|
resolution: StringResolutionResult,
|
|
output_dir: Path,
|
|
prefix: str,
|
|
) -> tuple[Path, Dict[str, Any]]:
|
|
"""Serialize one deterministic summary row per input relationship."""
|
|
columns = [
|
|
"canonical_target_id",
|
|
"ortholog_reference_id",
|
|
"raw_query_id",
|
|
"candidate_gene_id",
|
|
"selected_query_id",
|
|
"string_id",
|
|
"string_preferred_name",
|
|
"mapping_status",
|
|
"method",
|
|
"evidence_source",
|
|
"relationship_source",
|
|
"species",
|
|
"taxon_id",
|
|
"reason",
|
|
]
|
|
output_file = output_dir / f"{prefix}_id_resolution.tsv"
|
|
with output_file.open("w", encoding="utf-8") as handle:
|
|
handle.write("\t".join(columns) + "\n")
|
|
for row in resolution.rows:
|
|
values = [
|
|
str(getattr(row, column)).replace("\t", " ").replace("\n", " ")
|
|
for column in columns
|
|
]
|
|
handle.write("\t".join(values) + "\n")
|
|
return output_file, dict(resolution.metrics)
|
|
|
|
def analyze_network(
|
|
self,
|
|
species: str,
|
|
required_score: int = 200,
|
|
*,
|
|
origins: List[PpiQueryOrigin],
|
|
canonical_target_ids: List[str],
|
|
id_map_file: Path | None = None,
|
|
normalization_species_label: str = "",
|
|
) -> Dict:
|
|
"""分析互作网络
|
|
|
|
Returns:
|
|
Dict with keys: interactions, stats, string_url
|
|
"""
|
|
api = self.string_api_factory(species)
|
|
resolution = self.resolve_query_ids(
|
|
origins,
|
|
species,
|
|
id_map_file,
|
|
api=api,
|
|
canonical_target_ids=canonical_target_ids,
|
|
normalization_species_label=normalization_species_label,
|
|
)
|
|
network_query_ids = resolution.selected_string_ids
|
|
if resolution.status == "id_resolution_failure":
|
|
return {
|
|
"status": "id_resolution_failure",
|
|
"issue_code": resolution.issue_code,
|
|
"interactions": [],
|
|
"stats": {},
|
|
"string_url": "",
|
|
"id_resolution": resolution,
|
|
}
|
|
logger.info(
|
|
f"Querying STRING for {len(network_query_ids)} genes (species: {species})"
|
|
)
|
|
|
|
network_data = api.get_network(network_query_ids, required_score)
|
|
|
|
# 生成STRING网页链接 (无论是否有网络数据都生成)
|
|
string_url = generate_string_url(
|
|
network_query_ids, api.species_id, required_score
|
|
)
|
|
logger.info(f"STRING URL: {string_url}")
|
|
|
|
if not network_data:
|
|
logger.info("STRING returned a valid empty network")
|
|
return {
|
|
"interactions": [],
|
|
"stats": {
|
|
"input_genes": len(network_query_ids),
|
|
"genes_in_network": 0,
|
|
"total_interactions": 0,
|
|
"avg_score": 0,
|
|
"high_confidence": 0,
|
|
"medium_confidence": 0,
|
|
"low_confidence": 0,
|
|
"network_status": "empty_network",
|
|
},
|
|
"string_url": string_url,
|
|
"id_resolution": resolution,
|
|
}
|
|
|
|
# 解析互作数据
|
|
interactions = []
|
|
proteins_in_network = set()
|
|
|
|
for interaction in network_data:
|
|
if isinstance(interaction, dict):
|
|
p1 = interaction.get("preferredName_A", "")
|
|
p2 = interaction.get("preferredName_B", "")
|
|
score = interaction.get("score", 0)
|
|
|
|
interactions.append({
|
|
"protein1": p1,
|
|
"protein2": p2,
|
|
"score": score
|
|
})
|
|
|
|
proteins_in_network.add(p1)
|
|
proteins_in_network.add(p2)
|
|
|
|
# 统计
|
|
stats = {
|
|
"input_genes": len(network_query_ids),
|
|
"genes_in_network": len(proteins_in_network),
|
|
"total_interactions": len(interactions),
|
|
"avg_score": sum(i["score"] for i in interactions) / len(interactions) if interactions else 0,
|
|
"high_confidence": sum(1 for i in interactions if i["score"] >= 0.7),
|
|
"medium_confidence": sum(1 for i in interactions if 0.4 <= i["score"] < 0.7),
|
|
"low_confidence": sum(1 for i in interactions if i["score"] < 0.4),
|
|
"network_status": "network_detected",
|
|
}
|
|
|
|
logger.info(f"Network: {stats['genes_in_network']}/{stats['input_genes']} genes, "
|
|
f"{stats['total_interactions']} interactions")
|
|
|
|
return {
|
|
"interactions": interactions,
|
|
"stats": stats,
|
|
"string_url": string_url,
|
|
"id_resolution": resolution,
|
|
}
|
|
|
|
def save_network(self, interactions: List[Dict], output_file: Path) -> Path:
|
|
"""保存网络数据"""
|
|
df = pd.DataFrame(interactions, columns=["protein1", "protein2", "score"])
|
|
df.to_csv(output_file, sep='\t', index=False)
|
|
logger.info(f"Network saved: {output_file}")
|
|
return output_file
|
|
|
|
def run(
|
|
self,
|
|
gene_ids_file: Path,
|
|
species: str,
|
|
prefix: str,
|
|
output_dir: Path,
|
|
required_score: int = 200,
|
|
mapping_file: Optional[Path] = None,
|
|
id_map_file: Optional[Path] = None,
|
|
normalization_species_label: str = "",
|
|
ppi_provenance: Optional[Dict[str, Any]] = None,
|
|
) -> Dict:
|
|
"""运行PPI分析
|
|
|
|
Args:
|
|
gene_ids_file: 基因ID列表文件
|
|
species: 目标物种代码 (用于STRING查询)
|
|
prefix: 输出前缀
|
|
output_dir: 输出目录
|
|
required_score: 置信度阈值
|
|
mapping_file: 同源映射文件 (可选，用于非模式植物)
|
|
id_map_file: 当前任务/物种的本地ID证据文件 (可选)
|
|
normalization_species_label: 仅用于本地ID规范化规则，不作为STRING查询身份
|
|
"""
|
|
self.start_time = datetime.now()
|
|
|
|
logger.info(f"Starting PPI analysis for {prefix}")
|
|
logger.info(f"Input: {gene_ids_file}")
|
|
logger.info(f"Species: {species}")
|
|
logger.info(f"Output: {output_dir}")
|
|
|
|
# 创建输出目录
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 加载基因ID
|
|
gene_ids = self.load_gene_ids(gene_ids_file)
|
|
logger.info(f"Loaded {len(gene_ids)} gene IDs")
|
|
|
|
if len(gene_ids) == 0:
|
|
logger.error("No gene IDs loaded")
|
|
return self._generate_failed_manifest(
|
|
prefix, output_dir, "No gene IDs", ppi_provenance=ppi_provenance
|
|
)
|
|
|
|
# 加载同源映射 (如果有)
|
|
ortholog_mode = mapping_file is not None
|
|
self.ortholog_mapping = None
|
|
if ortholog_mode:
|
|
if not mapping_file.exists():
|
|
return self._generate_failed_manifest(
|
|
prefix,
|
|
output_dir,
|
|
f"Ortholog mapping file does not exist: {mapping_file}",
|
|
ppi_provenance=ppi_provenance,
|
|
)
|
|
self.load_ortholog_mapping(mapping_file)
|
|
|
|
if ortholog_mode:
|
|
requested_members = set(gene_ids)
|
|
origins = [
|
|
origin
|
|
for origin in (self.ortholog_mapping or [])
|
|
if origin.canonical_target_id in requested_members
|
|
]
|
|
mapped_members = {origin.canonical_target_id for origin in origins}
|
|
if len(mapped_members) < len(requested_members):
|
|
logger.warning(
|
|
"%s/%s canonical members have no Ortholog relationship",
|
|
len(requested_members) - len(mapped_members),
|
|
len(requested_members),
|
|
)
|
|
else:
|
|
origins = [
|
|
PpiQueryOrigin(
|
|
canonical_target_id=gene_id,
|
|
ortholog_reference_id="",
|
|
raw_query_id=gene_id,
|
|
relationship_source=str(gene_ids_file),
|
|
)
|
|
for gene_id in gene_ids
|
|
]
|
|
result = self.analyze_network(
|
|
species,
|
|
required_score,
|
|
origins=origins,
|
|
id_map_file=id_map_file,
|
|
canonical_target_ids=gene_ids,
|
|
normalization_species_label=normalization_species_label,
|
|
)
|
|
resolution = result["id_resolution"]
|
|
id_resolution_file, id_resolution_metrics = (
|
|
self.write_string_resolution_report(resolution, output_dir, prefix)
|
|
)
|
|
if result.get("status") == "id_resolution_failure":
|
|
return self._generate_failed_manifest(
|
|
prefix,
|
|
output_dir,
|
|
"No identifiers resolved unambiguously in STRING",
|
|
issue_code=str(result.get("issue_code") or ""),
|
|
metrics={"id_resolution": id_resolution_metrics},
|
|
artifacts={"id_resolution": self._record_file(id_resolution_file, "tsv")},
|
|
ppi_provenance=ppi_provenance,
|
|
)
|
|
query_ids = resolution.selected_string_ids
|
|
|
|
interactions = result["interactions"]
|
|
stats = result["stats"]
|
|
string_url = result.get("string_url", "")
|
|
|
|
# 保存网络数据
|
|
network_file = output_dir / f"{prefix}_network.tsv"
|
|
self.save_network(interactions, network_file)
|
|
|
|
# 保存STRING URL到文件
|
|
if string_url:
|
|
url_file = output_dir / f"{prefix}_url.txt"
|
|
with open(url_file, 'w') as f:
|
|
f.write("# STRING Protein Interaction Network\n\n")
|
|
f.write(f"**Query**: {prefix}\n")
|
|
f.write(f"**Species**: {species}\n")
|
|
f.write(f"**Genes**: {len(query_ids)}\n\n")
|
|
f.write(f"View online: {string_url}\n")
|
|
logger.info(f"STRING URL saved: {url_file}")
|
|
|
|
# 生成报告
|
|
report_path = self._generate_report(stats, prefix, output_dir, species, string_url)
|
|
|
|
# 生成manifest
|
|
manifest = self._generate_manifest(
|
|
gene_ids_file,
|
|
species,
|
|
prefix,
|
|
output_dir,
|
|
stats,
|
|
network_file,
|
|
report_path,
|
|
string_url,
|
|
id_resolution_file=id_resolution_file,
|
|
id_resolution_metrics=id_resolution_metrics,
|
|
normalization_species_label=normalization_species_label,
|
|
ppi_provenance=ppi_provenance,
|
|
)
|
|
|
|
self.end_time = datetime.now()
|
|
elapsed = (self.end_time - self.start_time).total_seconds()
|
|
logger.info(f"PPI analysis completed in {elapsed:.1f}s")
|
|
|
|
return manifest
|
|
|
|
def _generate_report(self, stats: Dict, prefix: str, output_dir: Path,
|
|
species: str, string_url: str = "") -> Path:
|
|
"""生成Markdown报告"""
|
|
report_path = output_dir / f"{prefix}_report.md"
|
|
|
|
with open(report_path, 'w') as f:
|
|
f.write(f"# PPI Network Analysis Report - {prefix}\n\n")
|
|
f.write(f"**Generated**: {datetime.now().isoformat()}\n")
|
|
f.write(f"**Species**: {species}\n")
|
|
f.write(f"**Data Source**: STRING Database\n\n")
|
|
|
|
f.write("## Network Statistics\n\n")
|
|
f.write(f"- **Input Genes**: {stats.get('input_genes', 0)}\n")
|
|
f.write(f"- **Genes in Network**: {stats.get('genes_in_network', 0)}\n")
|
|
f.write(f"- **Total Interactions**: {stats.get('total_interactions', 0)}\n")
|
|
f.write(f"- **Average Score**: {stats.get('avg_score', 0):.3f}\n")
|
|
f.write(f"- **High Confidence (≥0.7)**: {stats.get('high_confidence', 0)}\n")
|
|
f.write(f"- **Medium Confidence (0.4-0.7)**: {stats.get('medium_confidence', 0)}\n")
|
|
f.write(f"- **Low Confidence (<0.4)**: {stats.get('low_confidence', 0)}\n\n")
|
|
|
|
if string_url:
|
|
f.write("## STRING Network\n\n")
|
|
f.write(f"[View Interactive Network]({string_url})\n\n")
|
|
|
|
f.write("## Output Files\n\n")
|
|
f.write(f"- `{prefix}_network.tsv` - Interaction network data\n")
|
|
f.write(f"- `{prefix}_report.md` - This report\n")
|
|
|
|
return report_path
|
|
|
|
def _generate_manifest(self, input_file: Path, species: str, prefix: str,
|
|
output_dir: Path, stats: Dict, network_file: Path,
|
|
report: Path, string_url: str = "",
|
|
id_resolution_file: Optional[Path] = None,
|
|
id_resolution_metrics: Optional[Dict] = None,
|
|
normalization_species_label: str = "",
|
|
ppi_provenance: Optional[Dict[str, Any]] = None) -> Dict:
|
|
"""生成manifest"""
|
|
self.end_time = datetime.now()
|
|
elapsed = (self.end_time - self.start_time).total_seconds()
|
|
|
|
artifacts = {}
|
|
|
|
if network_file and network_file.exists():
|
|
artifacts["network"] = self._record_file(network_file, "tsv")
|
|
|
|
if report and report.exists():
|
|
artifacts["report"] = self._record_file(report, "markdown")
|
|
if id_resolution_file and id_resolution_file.exists():
|
|
artifacts["id_resolution"] = self._record_file(id_resolution_file, "tsv")
|
|
|
|
# 添加URL文件到artifacts
|
|
url_file = output_dir / f"{prefix}_url.txt"
|
|
if url_file.exists():
|
|
artifacts["string_url"] = self._record_file(url_file, "txt")
|
|
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "ppi",
|
|
"status": "success",
|
|
"prefix": prefix,
|
|
"timestamp": self.end_time.isoformat(),
|
|
"elapsed_sec": round(elapsed, 2),
|
|
"input": {"gene_ids": str(input_file), "species": species},
|
|
"parameters": {
|
|
"string_taxon_id": StringAPI.resolve_species_id(species),
|
|
"normalization_species_label": str(normalization_species_label or species).strip(),
|
|
},
|
|
"metrics": {
|
|
**stats,
|
|
"id_resolution": id_resolution_metrics or {},
|
|
},
|
|
"artifacts": artifacts,
|
|
"string_url": string_url,
|
|
"ppi_provenance": dict(ppi_provenance or {}),
|
|
}
|
|
|
|
manifest_path = output_dir / f"{prefix}.manifest.json"
|
|
with open(manifest_path, 'w') as f:
|
|
json.dump(manifest, f, indent=2)
|
|
|
|
manifest["manifest_path"] = str(manifest_path)
|
|
return manifest
|
|
|
|
def _generate_failed_manifest(
|
|
self,
|
|
prefix: str,
|
|
output_dir: Path,
|
|
message: str,
|
|
*,
|
|
issue_code: str = "ppi_execution_failed",
|
|
metrics: Optional[Dict[str, Any]] = None,
|
|
artifacts: Optional[Dict[str, Any]] = None,
|
|
ppi_provenance: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
manifest = {
|
|
"manifest_version": "1.0",
|
|
"stage": "ppi",
|
|
"status": "failed",
|
|
"prefix": prefix,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"message": message,
|
|
"issue_code": issue_code,
|
|
"metrics": dict(metrics or {}),
|
|
"artifacts": dict(artifacts or {}),
|
|
"ppi_provenance": dict(ppi_provenance or {}),
|
|
}
|
|
manifest_path = output_dir / f"{prefix}.manifest.json"
|
|
manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
|
|
return {**manifest, "manifest_path": str(manifest_path)}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="PPI Skill - Protein-Protein Interaction Analysis")
|
|
parser.add_argument("-i", "--gene-ids", help="Gene ID list file")
|
|
parser.add_argument("-s", "--species", default="ath", help="Species code (ath/osa/zma)")
|
|
parser.add_argument(
|
|
"--normalization-species-label",
|
|
default="",
|
|
help="Local species label used only for gene-ID normalization rules",
|
|
)
|
|
parser.add_argument("-p", "--prefix", required=True, help="Output prefix")
|
|
parser.add_argument("-o", "--output", required=True, help="Output directory")
|
|
parser.add_argument("--score", type=int, default=200, help="Confidence threshold (0-1000)")
|
|
parser.add_argument("--mapping", help="Ortholog mapping file (for non-model species)")
|
|
parser.add_argument("--id-map", help="Species-matched local PEP/CDS ID evidence TSV")
|
|
parser.add_argument("--provenance-json", help="Authoritative PPI provenance JSON from preflight/backend")
|
|
parser.add_argument("--config", help="Config file path")
|
|
parser.add_argument("--demo", action="store_true", help="Demo mode")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Demo模式
|
|
if args.demo:
|
|
output_path = Path(args.output)
|
|
output_path.mkdir(parents=True, exist_ok=True)
|
|
demo_genes = output_path / "demo_genes.txt"
|
|
with open(demo_genes, 'w') as f:
|
|
f.write("AT1G01010\nAT1G01020\nAT1G01030\n")
|
|
args.gene_ids = str(demo_genes)
|
|
|
|
if not args.gene_ids:
|
|
parser.error("--gene-ids is required (or use --demo)")
|
|
|
|
# 运行技能
|
|
skill = PpiSkill(config_path=args.config)
|
|
provenance = {}
|
|
if args.provenance_json:
|
|
provenance = json.loads(Path(args.provenance_json).read_text(encoding="utf-8"))
|
|
try:
|
|
result = skill.run(
|
|
gene_ids_file=Path(args.gene_ids),
|
|
species=args.species,
|
|
prefix=args.prefix,
|
|
output_dir=Path(args.output),
|
|
required_score=args.score,
|
|
mapping_file=Path(args.mapping) if args.mapping else None,
|
|
id_map_file=Path(args.id_map) if args.id_map else None,
|
|
normalization_species_label=args.normalization_species_label,
|
|
ppi_provenance=provenance,
|
|
)
|
|
except Exception as exc:
|
|
result = skill._generate_failed_manifest(
|
|
args.prefix, Path(args.output), str(exc), ppi_provenance=provenance
|
|
)
|
|
|
|
if result.get("status") == "failed":
|
|
print(f"\n❌ PPI analysis failed: {result.get('message')}")
|
|
sys.exit(1)
|
|
|
|
print(f"\n✅ PPI analysis completed!")
|
|
print(f" Manifest: {result.get('manifest_path', 'N/A')}")
|
|
print(f" Genes in network: {result['metrics'].get('genes_in_network', 0)}/{result['metrics'].get('input_genes', 0)}")
|
|
print(f" Interactions: {result['metrics'].get('total_interactions', 0)}")
|
|
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|