289 lines
9.5 KiB
Markdown
Executable File
289 lines
9.5 KiB
Markdown
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/`
|
|
|
|
**处理**:
|
|
```python
|
|
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`（多个）
|
|
|
|
**处理**:
|
|
```python
|
|
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文件, 配置文件等 │
|
|
├─────────────────────────────────────────┤
|
|
│ [开始上传并处理] │
|
|
└─────────────────────────────────────────┘
|
|
```
|
|
|
|
### 参数配置
|
|
|
|
```yaml
|
|
# 自动从文件夹结构生成
|
|
analysis:
|
|
target_species: "Ft" # 从target/文件夹名获取
|
|
target_prefix: "Ft.SPL" # 用户输入
|
|
reference_species: # 从references/子文件夹获取
|
|
- "Ath"
|
|
- "Os"
|
|
- "Sl"
|
|
gene_family: "SPL" # 从prefix解析
|
|
```
|
|
|
|
## 实现代码
|
|
|
|
### 文件夹扫描
|
|
|
|
```python
|
|
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
|
|
```
|
|
|
|
### 处理流程
|
|
|
|
```python
|
|
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)") |