244 lines
7.9 KiB
Python
Executable File
244 lines
7.9 KiB
Python
Executable File
"""
|
||
Pfam API 客户端
|
||
|
||
Pfam 是一个蛋白质家族数据库，基于隐马尔可夫模型 (HMM)
|
||
URL: https://pfam.xfam.org/
|
||
API: https://xfam.wordpress.com/2020/07/08/pfam-33-1-is-available/
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
from typing import Dict, List, Optional, Any, Tuple
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass
|
||
class PfamEntry:
|
||
"""Pfam 条目"""
|
||
accession: str
|
||
name: str
|
||
description: str
|
||
clan: Optional[str] = None
|
||
domain_architecture: Optional[List[str]] = None
|
||
|
||
|
||
class PfamClient:
|
||
"""
|
||
Pfam 数据库客户端
|
||
|
||
用于查询蛋白质家族信息、HMM 模型、结构域架构
|
||
"""
|
||
|
||
# Pfam REST API (通过 EBI)
|
||
BASE_URL = "https://www.ebi.ac.uk/interpro/api"
|
||
PFAM_URL = "https://pfam.xfam.org"
|
||
|
||
def __init__(self, rate_limit: float = 0.5):
|
||
"""
|
||
初始化客户端
|
||
|
||
Args:
|
||
rate_limit: 请求间隔（秒）
|
||
"""
|
||
self.rate_limit = rate_limit
|
||
self.last_request_time = 0
|
||
self.session = requests.Session()
|
||
self.session.headers.update({
|
||
'User-Agent': 'bbav20311/1.0 (Research Tool)',
|
||
'Accept': 'application/json'
|
||
})
|
||
|
||
def _rate_limited_request(self, url: str, **kwargs) -> requests.Response:
|
||
"""带速率限制的请求"""
|
||
elapsed = time.time() - self.last_request_time
|
||
if elapsed < self.rate_limit:
|
||
time.sleep(self.rate_limit - elapsed)
|
||
|
||
response = self.session.get(url, **kwargs)
|
||
self.last_request_time = time.time()
|
||
|
||
return response
|
||
|
||
def search_protein(self, protein_id: str) -> List[PfamEntry]:
|
||
"""
|
||
搜索蛋白质的结构域
|
||
|
||
Args:
|
||
protein_id: 蛋白质 ID（UniProt ID）
|
||
|
||
Returns:
|
||
Pfam 条目列表
|
||
"""
|
||
url = f"{self.BASE_URL}/entry/pfam/protein/uniprot/{protein_id}"
|
||
|
||
try:
|
||
response = self._rate_limited_request(url)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
entries = []
|
||
for result in data.get('results', []):
|
||
entry = PfamEntry(
|
||
accession=result.get('metadata', {}).get('accession', ''),
|
||
name=result.get('metadata', {}).get('name', ''),
|
||
description=result.get('metadata', {}).get('description', ''),
|
||
clan=result.get('metadata', {}).get('clan', {}).get('accession') if result.get('metadata', {}).get('clan') else None
|
||
)
|
||
entries.append(entry)
|
||
|
||
return entries
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"[WARN] Pfam API request failed: {e}")
|
||
return []
|
||
|
||
def get_family_info(self, pfam_acc: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
获取 Pfam 家族详细信息
|
||
|
||
Args:
|
||
pfam_acc: Pfam 登录号（如 PF03110）
|
||
|
||
Returns:
|
||
家族详细信息
|
||
"""
|
||
url = f"{self.BASE_URL}/entry/pfam/{pfam_acc}"
|
||
|
||
try:
|
||
response = self._rate_limited_request(url)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
metadata = data.get('metadata', {})
|
||
return {
|
||
'accession': metadata.get('accession'),
|
||
'name': metadata.get('name'),
|
||
'description': metadata.get('description'),
|
||
'clan': metadata.get('clan', {}).get('accession') if metadata.get('clan') else None,
|
||
'clan_name': metadata.get('clan', {}).get('name') if metadata.get('clan') else None,
|
||
'go_terms': metadata.get('go_terms', []),
|
||
'interpro': metadata.get('interpro', {}),
|
||
'protein_count': data.get('proteins', {}).get('total', 0),
|
||
'structure_count': data.get('structures', {}).get('total', 0),
|
||
}
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"[WARN] Pfam API request failed: {e}")
|
||
return None
|
||
|
||
def search_by_keyword(self, keyword: str, page_size: int = 20) -> List[Dict[str, Any]]:
|
||
"""
|
||
按关键词搜索 Pfam 家族
|
||
|
||
Args:
|
||
keyword: 搜索关键词
|
||
page_size: 每页结果数
|
||
|
||
Returns:
|
||
搜索结果列表
|
||
"""
|
||
url = f"{self.BASE_URL}/entry/pfam"
|
||
params = {
|
||
'search': keyword,
|
||
'page_size': page_size,
|
||
}
|
||
|
||
try:
|
||
response = self._rate_limited_request(url, params=params)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
results = []
|
||
for result in data.get('results', []):
|
||
metadata = result.get('metadata', {})
|
||
results.append({
|
||
'accession': metadata.get('accession'),
|
||
'name': metadata.get('name'),
|
||
'description': metadata.get('description'),
|
||
'clan': metadata.get('clan', {}).get('accession') if metadata.get('clan') else None,
|
||
})
|
||
|
||
return results
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"[WARN] Pfam API request failed: {e}")
|
||
return []
|
||
|
||
def get_hmm_model(self, pfam_acc: str) -> Optional[str]:
|
||
"""
|
||
获取 HMM 模型（如果可用）
|
||
|
||
Args:
|
||
pfam_acc: Pfam 登录号
|
||
|
||
Returns:
|
||
HMM 模型字符串
|
||
"""
|
||
# Pfam HMM 模型通常需要从 FTP 或特定端点下载
|
||
# 这里返回下载链接
|
||
return f"https://pfam.xfam.org/family/{pfam_acc}/hmm"
|
||
|
||
def get_clan_info(self, clan_acc: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
获取 Clan（超家族）信息
|
||
|
||
Args:
|
||
clan_acc: Clan 登录号（如 CL0016）
|
||
|
||
Returns:
|
||
Clan 信息
|
||
"""
|
||
url = f"{self.BASE_URL}/entry/pfam/clan/{clan_acc}"
|
||
|
||
try:
|
||
response = self._rate_limited_request(url)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
return {
|
||
'accession': data.get('metadata', {}).get('accession'),
|
||
'name': data.get('metadata', {}).get('name'),
|
||
'description': data.get('metadata', {}).get('description'),
|
||
'member_families': [f.get('accession') for f in data.get('members', [])],
|
||
}
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"[WARN] Pfam API request failed: {e}")
|
||
return None
|
||
|
||
def analyze_protein_domains(self, protein_sequence: str) -> Dict[str, Any]:
|
||
"""
|
||
分析蛋白质序列的结构域
|
||
|
||
注意：这需要提交到 Pfam 搜索（可能需要网页界面）
|
||
|
||
Args:
|
||
protein_sequence: 蛋白质序列
|
||
|
||
Returns:
|
||
结构域分析结果
|
||
"""
|
||
# Pfam 序列搜索通常通过网页表单或 InterProScan
|
||
# 这里返回搜索链接
|
||
return {
|
||
'search_url': 'https://www.ebi.ac.uk/jdispatcher/pfa/iprscan5',
|
||
'note': 'Use InterProScan for sequence analysis',
|
||
'sequence_length': len(protein_sequence),
|
||
}
|
||
|
||
|
||
# 使用示例
|
||
if __name__ == "__main__":
|
||
client = PfamClient()
|
||
|
||
# 搜索 SBP 家族
|
||
results = client.search_by_keyword("SBP")
|
||
print(f"Found {len(results)} Pfam families")
|
||
|
||
# 获取 PF03110 (SBP) 详细信息
|
||
info = client.get_family_info("PF03110")
|
||
if info:
|
||
print(f"Family: {info['name']}")
|
||
print(f"Description: {info['description']}")
|
||
print(f"Protein count: {info['protein_count']}")
|