663 纳米孔宏基因组实时分析¶

# WIMP流程（2024年已停止服务）
MinION测序 → EPI2ME Agent（云端）→ Centrifuge分类 → 实时结果
问题：需要稳定网络、不灵活、数据库固定

# 安装wf-metagenomics（EPI2ME Labs工作流）
# 方法1：通过Nextflow运行
nextflow run epi2me-labs/wf-metagenomics \
  --fastq input_reads/ \          # 输入：fastq文件目录
  --classifier kraken2 \          # 使用Kraken2分类器
  --database_set Standard-8 \     # 标准数据库（8GB）
  --out_dir results/ \            # 输出目录
  --real_time true                # 启用实时模式

# 方法2：实时监控模式（边测序边分析）
nextflow run epi2me-labs/wf-metagenomics \
  --fastq /data/minion_output/ \  # MinION输出目录
  --watch_path true \             # 监控目录，新文件自动分析
  --classifier kraken2 \          # Kraken2分类
  --database_set Standard-8 \     # 数据库
  --amr true \                    # 同时检测抗性基因
  --out_dir realtime_results/     # 输出目录

# 构建自定义Kraken2数据库
kraken2-build --download-taxonomy --db custom_db/  # 下载分类信息

# 添加目标物种的基因组
kraken2-build --add-to-library pathogen_genomes.fna \
  --db custom_db/  # 添加自定义基因组

# 构建数据库
kraken2-build --build --db custom_db/ \
  --threads 16  # 使用16线程

# 使用自定义数据库运行
nextflow run epi2me-labs/wf-metagenomics \
  --fastq reads/ \
  --database custom_db/ \         # 使用自定义数据库
  --taxonomy custom_db/taxonomy/  # 分类信息路径
  --out_dir results/

# MARTi安装（开源工具）
git clone https://github.com/richardmleggett/MARTi.git  # 克隆仓库
cd MARTi  # 进入目录

# 配置MARTi
# MARTi支持多种分类器：BLAST、Centrifuge、Kraken2
cat > config.yaml << 'EOF'
classifier: kraken2               # 使用Kraken2作为分类器
database: /path/to/kraken2_db     # 数据库路径
input_dir: /data/minion_output    # MinION输出目录
output_dir: /data/marti_results   # 结果输出目录
update_interval: 30               # 每30秒更新一次结果
min_quality: 7                    # 最低质量过滤
min_length: 200                   # 最短读长过滤
EOF

# 运行实时分析
java -jar MARTi.jar \
  --config config.yaml \          # 配置文件
  --realtime                      # 实时模式

# MARTi特点：
# - 支持BLAST/Centrifuge/Kraken2多种分类器
# - 交互式可视化界面
# - 用户可自定义参数和数据库
# - 不需要网络连接（离线运行）

# 自建实时纳米孔宏基因组分析脚本
import os  # 系统操作
import time  # 时间控制
import subprocess  # 调用外部程序
from watchdog.observers import Observer  # 文件系统监控
from watchdog.events import FileSystemEventHandler  # 事件处理
import pandas as pd  # 数据处理

class FastqHandler(FileSystemEventHandler):
    """监控新的fastq文件并自动分析"""

    def __init__(self, db_path, output_dir):
        self.db_path = db_path  # Kraken2数据库路径
        self.output_dir = output_dir  # 输出目录
        self.total_reads = 0  # 已分析读段总数
        self.taxa_counts = {}  # 物种计数字典

    def on_created(self, event):
        """新文件创建时触发"""
        if event.src_path.endswith('.fastq') or event.src_path.endswith('.fastq.gz'):
            print(f"[{time.strftime('%H:%M:%S')}] 发现新文件: {os.path.basename(event.src_path)}")
            self.classify_reads(event.src_path)  # 分类新读段
            self.update_report()  # 更新报告

    def classify_reads(self, fastq_path):
        """使用Kraken2分类读段"""
        output_file = os.path.join(self.output_dir, "kraken2_output.txt")
        report_file = os.path.join(self.output_dir, "kraken2_report.txt")

        cmd = [
            "kraken2",
            "--db", self.db_path,      # 数据库
            "--output", output_file,    # 逐条分类结果
            "--report", report_file,    # 汇总报告
            "--threads", "8",           # 线程数
            "--minimum-hit-groups", "3",  # 最少命中组数
            fastq_path                  # 输入文件
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        print(f"  分类完成: {result.stderr.strip()}")

        # 解析报告
        self.parse_report(report_file)

    def parse_report(self, report_file):
        """解析Kraken2报告"""
        with open(report_file) as f:
            for line in f:
                parts = line.strip().split('\t')
                if len(parts) >= 6:
                    pct = float(parts[0])  # 百分比
                    count = int(parts[1])  # 读段数
                    rank = parts[3].strip()  # 分类级别
                    name = parts[5].strip()  # 物种名
                    if rank == 'S' and pct > 0.1:  # 物种级别且>0.1%
                        self.taxa_counts[name] = count

    def update_report(self):
        """更新实时报告"""
        if self.taxa_counts:
            df = pd.DataFrame.from_dict(self.taxa_counts,
                                         orient='index', columns=['reads'])
            df = df.sort_values('reads', ascending=False)
            df['percentage'] = df['reads'] / df['reads'].sum() * 100
            print("\n=== 当前Top10物种 ===")
            print(df.head(10).to_string())
            print("====================\n")

# 启动实时监控
def start_realtime_analysis(watch_dir, db_path, output_dir):
    """启动实时分析"""
    os.makedirs(output_dir, exist_ok=True)
    handler = FastqHandler(db_path, output_dir)  # 事件处理器
    observer = Observer()  # 文件系统观察者
    observer.schedule(handler, watch_dir, recursive=False)  # 监控目录
    observer.start()  # 开始监控
    print(f"开始监控目录: {watch_dir}")
    try:
        while True:
            time.sleep(10)  # 每10秒检查一次
    except KeyboardInterrupt:
        observer.stop()  # Ctrl+C停止
    observer.join()

# 使用
# start_realtime_analysis("/data/minion", "/db/kraken2", "/results")

# 使用ABRicate进行实时抗性基因检测
# wf-metagenomics内置AMR检测功能

# 独立运行ABRicate
abricate \
  --db card \                   # 使用CARD数据库
  --minid 80 \                  # 最低序列一致性80%
  --mincov 60 \                 # 最低覆盖度60%
  assembled_contigs.fa \        # 输入组装的contigs
  > amr_results.tab             # 输出结果

# 或直接对长读段运行
abricate \
  --db resfinder \              # 使用ResFinder数据库
  nanopore_reads.fastq \        # 纳米孔读段
  > amr_reads_results.tab       # 结果

# 纳米孔宏基因组分析流程
MinION测序 → 实时basecalling(Dorado)
  → 质控(NanoPlot/chopper)
  → 实时分类(Kraken2/wf-metagenomics)
  → AMR检测(ABRicate)
  → 可视化报告(MARTi)

# 关键工具
basecalling: Dorado (ONT官方，取代Guppy)
质控: NanoPlot + chopper/Nanofilt
分类: Kraken2 (推荐) / Centrifuge
AMR: ABRicate + CARD/ResFinder
组装: Flye (长读段组装器)
可视化: MARTi (2025) / EPI2ME Desktop

# 数据库选择
Standard-8: 8GB，基本细菌/古菌/病毒
Standard-16: 16GB，更全面
PlusPF: 含真核寄生虫和真菌
自定义: kraken2-build添加目标基因组

# 实时分析要点
1. 选小而精的数据库（减少内存+提速）
2. 设置最小读长过滤（>200bp）
3. 设置最小质量过滤（Q>7）
4. 每批次100-1000条reads更新一次结果