9.5 KiB
Executable File
9.5 KiB
Executable File
基于文件夹的多物种工作流设计
核心设计
通过物理文件夹结构明确区分目标物种和参考物种：
input/{job_id}/
├── target/ # 目标物种（必需，唯一）
│ ├── Ft.pep.fasta # 蛋白序列
│ ├── Ft.gff3 # 注释文件
│ └── Ft.genome.fasta # 基因组（可选）
│
├── references/ # 参考物种（可选，可多个）
│ ├── Ath/ # 拟南芥
│ │ ├── Ath.pep.fasta
│ │ └── Ath.gff3
│ ├── Os/ # 水稻
│ │ ├── Os.pep.fasta
│ │ └── Os.gff3
│ └── Sl/ # 番茄
│ ├── Sl.pep.fasta
│ └── Sl.gff3
│
└── optional/ # 其他可选文件
├── SBP.hmm # HMM文件
└── custom_config.yaml # 自定义配置
文件夹结构说明
| 文件夹 | 用途 | 数量限制 | 必需文件 |
|---|---|---|---|
target/ |
目标物种数据 | 1个物种 | *.pep.fasta 或 *.pep.fa |
references/ |
参考物种数据 | 0-N个物种 | 每个子文件夹需要 *.pep.fasta |
optional/ |
可选文件 | 任意 | 无 |
工作流程
阶段1: 数据准备 (genome_prepare)
输入: input/{job_id}/target/ 和 input/{job_id}/references/
处理:
def process_target_folder(target_dir: Path, output_dir: Path, prefix: str):
"""处理目标物种文件夹"""
# 查找蛋白文件
pep_files = list(target_dir.glob("*.pep.fasta")) + list(target_dir.glob("*.pep.fa"))
if not pep_files:
raise FileNotFoundError("目标物种文件夹缺少蛋白文件 (*.pep.fasta)")
# 处理目标物种
process_species(target_dir, output_dir, prefix, is_target=True)
def process_reference_folders(ref_dir: Path, output_dir: Path):
"""处理参考物种文件夹"""
references = []
for species_dir in ref_dir.iterdir():
if not species_dir.is_dir():
continue
species_code = species_dir.name # Ath, Os, Sl, etc.
# 检查必需文件
pep_files = list(species_dir.glob("*.pep.fasta")) + list(species_dir.glob("*.pep.fa"))
if not pep_files:
logger.warning(f"参考物种 {species_code} 缺少蛋白文件，跳过")
continue
# 处理参考物种
process_species(species_dir, output_dir, species_code, is_target=False)
references.append(species_code)
return references
输出:
input/{job_id}/
├── target/ # 原始目标物种数据
├── references/ # 原始参考物种数据
├── processed/ # 处理后的标准化数据
│ ├── target.pep.fasta # 目标物种蛋白（重命名）
│ ├── target.gff3
│ ├── Ath_ref.pep.fasta # 参考物种1（添加_ref后缀）
│ ├── Ath_ref.gff3
│ ├── Os_ref.pep.fasta # 参考物种2
│ ├── Os_ref.gff3
│ └── ...
└── genome_prepare.manifest.json
阶段2: 基因家族鉴定 (identify)
输入:
- 目标蛋白:
processed/target.pep.fasta - 参考蛋白:
processed/*_ref.pep.fasta（多个）
处理:
def run_identify_with_references(
target_pep: Path,
reference_peps: List[Path],
hmm_file: Optional[Path],
output_dir: Path
):
"""使用多参考物种进行基因家族鉴定"""
# 1. HMM搜索目标物种
hmm_results = run_hmmsearch(target_pep, hmm_file)
# 2. 合并参考物种构建BLAST数据库
if reference_peps:
combined_db = combine_references(reference_peps)
blast_db = build_blast_db(combined_db)
# 3. BLAST搜索
blast_results = run_blastp(target_pep, blast_db)
# 4. 分析参考物种来源
ref_analysis = analyze_reference_sources(blast_results)
logger.info(f"参考物种匹配情况: {ref_analysis}")
# 5. 整合结果
final_members = integrate_results(hmm_results, blast_results)
return final_members
def combine_references(reference_peps: List[Path]) -> Path:
"""合并多个参考物种的蛋白序列"""
combined = Path("combined_ref.pep.fa")
with open(combined, 'w') as out:
for ref_pep in reference_peps:
species_code = ref_pep.stem.replace('_ref', '') # Ath_ref -> Ath
with open(ref_pep) as f:
for line in f:
if line.startswith('>'):
# 添加物种标识
out.write(f">{species_code}|{line[1:]}")
else:
out.write(line)
return combined
输出:
final/
├── {prefix}.final_members.pep.fa # 目标物种基因家族成员
├── {prefix}.final_members.id
├── {prefix}.identify.manifest.json
├── {prefix}.identify_report.md
└── reference_analysis.json # 参考物种分析
阶段3: 进化分析 (phylogeny)
输入:
- 目标物种基因家族成员
- 参考物种同源基因
处理: 构建包含多物种的系统发育树
前端UI设计
上传界面
┌─────────────────────────────────────────┐
│ 📁 目标物种 (Target Species) │
│ [选择文件夹] 或 [拖拽文件夹到此处] │
│ 必需: 蛋白序列文件 (*.pep.fasta) │
│ 可选: 注释文件 (*.gff3), 基因组文件 │
├─────────────────────────────────────────┤
│ 📁 参考物种 (Reference Species) │
│ [添加参考物种] │
│ ├─ 📁 Ath (拟南芥) [删除] │
│ │ [选择文件夹] │
│ ├─ 📁 Os (水稻) [删除] │
│ │ [选择文件夹] │
│ └─ [+ 添加更多参考物种] │
│ 可选: 用于辅助鉴定的已知物种 │
├─────────────────────────────────────────┤
│ 📁 可选文件 (Optional) │
│ [选择文件] HMM文件, 配置文件等 │
├─────────────────────────────────────────┤
│ [开始上传并处理] │
└─────────────────────────────────────────┘
参数配置
# 自动从文件夹结构生成
analysis:
target_species: "Ft" # 从target/文件夹名获取
target_prefix: "Ft.SPL" # 用户输入
reference_species: # 从references/子文件夹获取
- "Ath"
- "Os"
- "Sl"
gene_family: "SPL" # 从prefix解析
实现代码
文件夹扫描
class FolderScanner:
"""扫描输入文件夹结构"""
def scan(self, input_dir: Path) -> Dict[str, Any]:
"""扫描文件夹，返回结构信息"""
result = {
"target": None,
"references": [],
"optional": []
}
# 扫描目标物种
target_dir = input_dir / "target"
if target_dir.exists():
result["target"] = self.scan_species_folder(target_dir, is_target=True)
# 扫描参考物种
ref_dir = input_dir / "references"
if ref_dir.exists():
for species_dir in ref_dir.iterdir():
if species_dir.is_dir():
species_info = self.scan_species_folder(species_dir, is_target=False)
result["references"].append(species_info)
# 扫描可选文件
opt_dir = input_dir / "optional"
if opt_dir.exists():
result["optional"] = list(opt_dir.iterdir())
return result
def scan_species_folder(self, species_dir: Path, is_target: bool) -> Dict[str, Any]:
"""扫描物种文件夹"""
species_code = species_dir.name
files = {
"code": species_code,
"is_target": is_target,
"pep": None,
"gff": None,
"genome": None,
"cds": None
}
for file in species_dir.iterdir():
if not file.is_file():
continue
name_lower = file.name.lower()
if 'pep' in name_lower or 'protein' in name_lower:
files["pep"] = file
elif 'cds' in name_lower:
files["cds"] = file
elif name_lower.endswith('.gff3') or name_lower.endswith('.gff'):
files["gff"] = file
elif 'genome' in name_lower or 'chromosome' in name_lower:
files["genome"] = file
return files
处理流程
def process_folder_structure(input_dir: Path, output_dir: Path, prefix: str):
"""处理文件夹结构"""
scanner = FolderScanner()
structure = scanner.scan(input_dir)
# 检查目标物种
if not structure["target"]:
raise ValueError("缺少目标物种文件夹: input/{job_id}/target/")
if not structure["target"]["pep"]:
raise ValueError("目标物种文件夹缺少蛋白文件 (*.pep.fasta)")