圖 1: 序列萃取概念圖。選擇機制示意圖,Top K 選擇器從堆疊中標示並提取最佳序列至輸出區。
序列萃取模組是 PowerBarcoder QC 流程的重要組成部分,負責從各處理階段萃取代表性序列,提供完整的序列追蹤功能。此模組已完全整合至統一的 QC 流程中,作為 QC 流程的最終步驟執行。
- 代表性序列收集:從多個處理階段(denoise、merge 等)收集代表性序列
- 推薦序列萃取:萃取統一的推薦序列 FASTA 檔案供前端下載
- TOP K ASV 選取:實作智慧 TOP K 演算法,允許使用者自定義保留的 ASV 序列數量
- 配對完整性保證:確保 r1+r2、r1+ref 等配對序列的完整性
- 標準化輸出:統一的 FASTA 格式與標頭命名規範
- 調試與追蹤支援:提供 Debug 模式用於序列追蹤,推薦序列用於生產模式
採用「統一序列萃取器 + 雙模式運作」的核心設計理念:
- 統一入口:QCSequenceExtractor 提供統一的序列萃取控制
- 雙模式運作:Debug 模式萃取所有處理步驟序列;生產模式僅輸出推薦序列
- 配置驅動:透過 config.top_k_sequences 和 enable_debug_extraction 控制萃取行為
- QC 流程整合:作為 QC 流程的一部分,與報告生成流程協調執行
- 模組獨立性:產出的檔案僅供前端觸發下載或開發期間除錯使用,不被其他模組依賴
QCSequenceExtractor 是統一序列萃取器,提供雙模式運作架構:
class QCSequenceExtractor:
"""
統一序列萃取器
根據 PRD 設計,提供 Debug 模式與生產模式的統一控制
"""
def __init__(self, config, enable_debug_extraction: bool = False):
"""
初始化統一序列萃取器(Phase 3 簡化版本)
Args:
config: PowerBarcoderConfig 配置物件
enable_debug_extraction: 是否啟用 debug 序列萃取
"""
self.config = config
self.enable_debug_extraction = enable_debug_extraction
# Phase 3 簡化:序列萃取和推薦序列總是啟用
top_k_raw = getattr(config, "top_k_sequences", 1)
# 確保 top_k_sequences 是整數類型
try:
self.top_k_sequences = int(top_k_raw) if top_k_raw is not None else 1
except (ValueError, TypeError):
self.top_k_sequences = 1統一萃取入口:
def extract_all_sequences(self, locus: str) -> bool:
"""
統一序列萃取入口(Phase 3 簡化版本)
Args:
locus: 基因座名稱
Returns:
萃取是否成功
"""
success = True
# Debug 模式:萃取所有處理步驟的序列
if self.enable_debug_extraction:
debug_success = self._extract_debug_sequences(locus)
success = success and debug_success
# Phase 3:推薦序列總是萃取
recommendation_success = self._generate_unified_recommendation_fasta(locus)
success = success and recommendation_success
return success模式特點:
- Debug 模式:
enable_debug_extraction=True時,萃取所有處理步驟的序列供開發除錯 - 生產模式:
enable_debug_extraction=False時,僅萃取推薦序列供前端下載 - 推薦序列:無論哪種模式都會萃取,存放於
{locus}_result/qcResult/{locus}_recommended_sequences.fas
新增的推薦序列萃取功能:
def _generate_unified_recommendation_fasta(self, locus: str) -> bool:
"""
生成包含所有推薦序列的統一 FASTA 檔案
Returns:
生成是否成功
"""
# 從 QCDataManager 取得樣本推薦資訊
qc_manager = QCDataManager(locus, self.config.resultDataPath)
samples = qc_manager.get_all_samples()
# 建立推薦序列輸出檔案
output_path = join_path(
self.config.resultDataPath, f"{locus}_result", "qcResult",
f"{locus}_recommended_sequences.fas"
)
# 萃取有推薦序列的樣本
with open(output_path, "w") as f:
for sample in samples:
if sample.get("recommendation_sequence"):
# Roadmap: 改為結構化欄位 (batch, locus, source, confidence, rule, origin, asv, abundance)
header = f">{sample['sample_name']}|src={sample['recommendation_source']}|conf={sample['recommendation_confidence']:.3f}"
f.write(f"{header}\n")
f.write(f"{sample['recommendation_sequence']}\n")seq_extractor.py 提供分層式的功能架構,主要用於 Debug 模式的詳細序列萃取:
- QC 整合層:
extract_and_combine_sequences_for_locus()主要入口點,由 QC 流程呼叫 - 樣本處理層:
combine_sample_sequences()和combine_all_samples()樣本級別處理 - 策略選擇層:根據檔案型態選擇不同的萃取策略
- 萃取執行層:實作具體的序列萃取邏輯
- 工具支援層:豐度解析、檔案讀取、序列格式化等輔助功能
序列萃取模組已完全整合至 QC 流程:
# QC 流程中的序列萃取調用
def run_locus_qc(self, locus: str):
# Step 1: 生成 QC 報告
generate_csv_report(
locus=locus,
destination=csv_report_path,
result_data_path=self.result_data_path
)
# Step 2: 統一序列萃取
sequence_extractor = QCSequenceExtractor(
config=self.config,
enable_debug_extraction=self.enable_debug_extraction
)
sequence_extractor.extract_all_sequences(locus)整合特點:
- 統一配置:使用相同的 config 和 top_k 參數
- 錯誤處理:與 QC 流程共享錯誤處理機制
- 日誌記錄:統一的日誌格式與輸出
- 執行順序:在 QC 報告生成後執行,確保資料可用性
序列萃取策略依原始資料形態(多序列單檔 vs 多檔單序列)分為兩種,僅保留一張主表(下表)避免重複;原兩段描述合併:
| 步驟 | 分類 | TOP K 單位 | 取樣策略 (K>0) | K=1 預期 | 配對完整性 |
|---|---|---|---|---|---|
| denoise_r1 / r2 | 單檔多序列 | 序列 | 依標頭豐度排序取前 K 條 | 各 1 條 | N/A |
| dada2_merged | 單檔多序列 | 序列 | 同上 | 1 條 merged | N/A |
| merger_ncatr1r2 | 單檔多序列 | 序列 | 同上 | 1 條 concat | N/A |
| merger_r1 / r2 | 多檔單序列 | 檔案 | 依檔名豐度排序取 K 檔→所有序列 | 各 1 檔 / 1 序列 | 保留檔案全部 |
| merger_merged | 多檔單序列 | 檔案 | 同上 | 1 檔 1 序列 | 保留檔案全部 |
| merger_r1ref / r2ref | 多檔單序列 | 檔案 | 同上 | 1 檔 2 序列 | r/ref 不拆分 |
| merger_aligned | 多檔單序列 | 檔案 (r1,r2 分組) | r1/r2 各自排序取 K 檔 | 2 檔 4 序列 | 不跨組拆分 |
| merger_aligned_mafft | 多檔單序列 | 檔案 | 過濾含 Align → 取 K 檔 | 1 檔 2 序列 | 保留檔案全部 |
重點:多檔單序列策略核心為「選檔案→取該檔全部序列」,不可跨檔混合。
推薦的使用方式是透過 QCSequenceExtractor,這提供了統一的序列萃取控制:
from src.qc.seq_extractor import QCSequenceExtractor
# 建立萃取器(生產模式 - 僅推薦序列)
extractor = QCSequenceExtractor(
config=config,
enable_debug_extraction=False
)
# 萃取特定 locus 的推薦序列
success = extractor.extract_all_sequences("rbcL")
# 建立萃取器(Debug 模式 - 所有序列 + 推薦序列)
debug_extractor = QCSequenceExtractor(
config=config,
enable_debug_extraction=True
)
# 萃取所有序列類型
success = debug_extractor.extract_all_sequences("rbcL")推薦的整合使用方式是透過 QC 流程,這樣可以確保序列萃取與 QC 報告生成的協調:
from src.qc.qc_pipeline import run_qc_pipeline
# 執行完整 QC 流程(包含序列萃取)
# 會自動建立 QCSequenceExtractor 並執行
success = run_qc_pipeline(config, top_k=3)如果只需要執行 Debug 模式的詳細序列萃取功能:
from src.qc.seq_extractor import extract_and_combine_sequences_for_locus
# 對特定 locus 執行序列萃取
extract_and_combine_sequences_for_locus(
result_data_path="/path/to/result",
locus="rbcL",
top_k=3
)from src.qc.seq_extractor import QCSequenceExtractor
# 使用統一萃取器進行批次處理
extractor = QCSequenceExtractor(config=config, enable_debug_extraction=False)
loci = ["rbcL", "ITS2", "trnLF"]
for locus in loci:
print(f"Processing locus: {locus}")
success = extractor.extract_all_sequences(locus)
print(f"Completed: {locus}, Success: {success}")# 配置中設定 TOP K
config.top_k_sequences = 5
# 建立萃取器並萃取(使用配置中的 TOP K)
extractor = QCSequenceExtractor(config=config, enable_debug_extraction=True)
extractor.extract_all_sequences("rbcL")
# 或使用原有方法直接指定 TOP K
extract_and_combine_sequences_for_locus(
result_data_path, "rbcL", top_k=5
)
# 保留所有序列(不進行 TOP K 選取)
extract_and_combine_sequences_for_locus(
result_data_path, "rbcL", top_k=0 # 或 top_k=None
)TOP K ASV 機制是 seq_extractor.py 的核心功能,針對多檔單序列型態的步驟(如 merger_r1ref、merger_r2ref、merger_aligned、merger_aligned_mafft)實作按檔案/ASV 的選取邏輯,以確保保留最具生物學意義的 ASV 檔案及其完整配對序列。
TOP K ASV 機制的設計遵循以下核心原則:
- 生物學意義優先:豐度高的 ASV 更具代表性,應優先保留
- 配對關係完整:配對序列(如 r1+ref)必須同時保留,不可拆分
- 處理策略區分:根據檔案型態採用不同的選取策略
- 向前相容性:支援
K=0或None保留全部序列/檔案 - 錯誤容忍性:豐度解析失敗時採用預設值,不中斷處理
步驟分類與處理邏輯:
| 處理步驟 | 檔案型態 | TOP K 應用邏輯 | 預期結果(topK=1 時) |
|---|---|---|---|
denoise_r1、denoise_r2 |
一檔多序列(多 ASV) | 按序列豐度選取 TOP K 條序列 | 1 條序列 |
dada2_merged |
一檔多序列(多 ASV) | 按序列豐度選取 TOP K 條序列 | 1 條序列 |
merger_ncatr1r2 |
一檔多序列(多 ASV) | 按序列豐度選取 TOP K 條序列 | 1 條序列 |
merger_r1、merger_r2 |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有序列 | 1 個檔案,通常含 1 條序列 |
merger_merged |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有序列 | 1 個檔案,通常含 1 條序列 |
merger_r1ref |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有配對序列 | 1 個檔案,通常含 2 條序列(r1+ref) |
merger_r2ref |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有配對序列 | 1 個檔案,通常含 2 條序列(r2+ref) |
merger_aligned |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有配對序列 | 1-2 個檔案,每檔通常含 2 條序列(r1/r2+ref) |
merger_aligned_mafft |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取含"Align"的 TOP K 檔案,提取所有序列 | 1 個檔案,通常含 2 條序列(r1+r2) |
重要說明:
- 多檔單序列步驟的核心邏輯是「選檔案,取所有序列」,而非「跨檔案選序列」
- 對於
topK=1,merger_r1ref步驟應產出 2 條序列(來自 1 個檔案),而非 1 條序列 - 配對序列(如 r1+ref)必須同時保留,不能拆分
- 豐度資訊從檔案名稱或序列標頭中提取(格式如
0.750_abundance_150)
TOP K ASV 機制是 seq_extractor.py 的核心功能,針對多檔單序列型態的步驟(如 merger_r1ref、merger_r2ref、merger_aligned、merger_aligned_mafft)實作按檔案/ASV 的選取邏輯,以確保保留最具生物學意義的 ASV 檔案及其完整配對序列。
設計遵循以下核心原則:
- 生物學意義優先:豐度高的 ASV 更具代表性,應優先保留
- 配對關係完整:配對序列(如 r1+ref)必須同時保留,不可拆分
- 處理策略區分:根據檔案型態採用不同的選取策略
- 向前相容性:支援
K=0或None保留全部序列/檔案 - 錯誤容忍性:豐度解析失敗時採用預設值,不中斷處理
| 處理步驟 | 檔案型態 | TOP K 應用邏輯 | 預期結果(topK=1 時) |
|---|---|---|---|
denoise_r1、denoise_r2 |
一檔多序列(多 ASV) | 按序列豐度選取 TOP K 條序列 | 1 條序列 |
dada2_merged |
一檔多序列(多 ASV) | 按序列豐度選取 TOP K 條序列 | 1 條序列 |
merger_ncatr1r2 |
一檔多序列(多 ASV) | 按序列豐度選取 TOP K 條序列 | 1 條序列 |
merger_r1、merger_r2 |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有序列 | 1 個檔案,通常含 1 條序列 |
merger_merged |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有序列 | 1 個檔案,通常含 1 條序列 |
merger_r1ref |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有配對序列 | 1 個檔案,通常含 2 條序列(r1+ref) |
merger_r2ref |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有配對序列 | 1 個檔案,通常含 2 條序列(r2+ref) |
merger_aligned |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取 TOP K 個檔案,提取每檔所有配對序列 | 1-2 個檔案,每檔通常含 2 條序列(r1/r2+ref) |
merger_aligned_mafft |
多檔單序列(每 ASV 一檔) | 按檔案豐度選取含"Align"的 TOP K 檔案,提取所有序列 | 1 個檔案,通常含 2 條序列(r1+r2) |
重要說明:
- 多檔單序列步驟的核心邏輯是「選檔案,取所有序列」,而非「跨檔案選序列」
- 對於
topK=1,merger_r1ref步驟應產出 2 條序列(來自 1 個檔案),而非 1 條序列 - 配對序列(如 r1+ref)必須同時保留,不能拆分
- 豐度資訊從檔案名稱或序列標頭中提取(格式如
0.750_abundance_150)
多檔單序列步驟需要從檔案名稱中識別豐度資訊以進行排序。支援的檔案名稱格式包括:
sample_01_0.750_abundance_150_r1.fas
萃取邏輯會掃描檔案名稱中的數值部分,尋找介於 0.0-1.0 之間的小數作為豐度值。當解析失敗時會返回 0.0 作為預設值,確保處理流程不中斷。
提取邏輯:
- 以底線分割檔案名稱字串
- 尋找符合小數格式且在
0.0-1.0範圍內的數值 - 驗證該數值為豐度百分比
- 解析失敗時返回
0.0
一檔多序列步驟需要從 FASTA 標頭中識別豐度資訊。支援的標頭格式包括:
>sample_01_r1_0.850_abundance_200>sample_01_0.750_abundance_150
萃取邏輯會移除 ">" 符號後,掃描標頭中的數值部分,使用與檔案名稱相同的邏輯提取豐度值。
針對多檔單序列步驟的核心演算法,處理階段包括:
- 檔案發現與篩選:掃描目錄中符合樣本模式的檔案,套用特定步驟的篩選規則
- 檔案豐度提取與排序:從檔案名稱提取豐度值,按豐度降序排序所有候選檔案
- TOP K 檔案選取:根據 top_k 參數選取豐度最高的 K 個檔案,當 top_k=0 時保留全部檔案
- 序列提取:從每個選中的檔案提取所有序列,維持檔案內序列的完整配對關係
- merger_aligned_mafft 步驟:只保留檔名包含 "Align" 的檔案
- 檔案豐度解析失敗:當無法提取豐度時使用 0.0 作為預設值,不中斷處理流程
- 日誌輸出:記錄檔案選取過程與序列提取統計,便於除錯和驗證
merger_aligned 步驟與其他多檔單序列步驟不同,需要實現更複雜的 TOP K 選取邏輯:
- 分別處理 r1 和 r2 檔案:在
mergeResult/merger/aligned目錄中,通常存在以_r1.fas和_r2.fas結尾的檔案 - 分別對 r1 和 r2 取 TOP K:對 r1 檔案和 r2 檔案各自排序並選取 TOP K 個檔案
- 合併所有序列:將選中的 r1 檔案和 r2 檔案中的所有序列合併
假設 top_k=1 且目錄中存在以下檔案:
mergeResult/merger/aligned/
├── sample_01_0.920_abundance_300_r1.fas (2 條序列: r1+ref)
├── sample_01_0.850_abundance_250_r1.fas (2 條序列: r1+ref)
├── sample_01_0.750_abundance_200_r1.fas (2 條序列: r1+ref)
├── sample_01_0.880_abundance_280_r2.fas (2 條序列: r2+ref)
├── sample_01_0.820_abundance_240_r2.fas (2 條序列: r2+ref)
└── sample_01_0.760_abundance_210_r2.fas (2 條序列: r2+ref)
處理流程:
- 分離 r1 和 r2 檔案:
- r1 檔案:
sample_01_0.920_abundance_300_r1.fas(豐度最高) - r2 檔案:
sample_01_0.880_abundance_280_r2.fas(豐度最高)
- r1 檔案:
- 選取 TOP 1:
- 選中 r1 檔案:
sample_01_0.920_abundance_300_r1.fas - 選中 r2 檔案:
sample_01_0.880_abundance_280_r2.fas
- 選中 r1 檔案:
- 萃取序列:
- 從 r1 檔案萃取 2 條序列(r1+ref)
- 從 r2 檔案萃取 2 條序列(r2+ref)
- 總計:4 條序列
對於 denoise_r1、denoise_r2、dada2_merged、merger_ncatr1r2 等一檔多序列步驟:
- 檔案定位:在指定目錄中尋找符合樣本模式的檔案,通常每個樣本只有一個檔案
- 序列讀取:從檔案中讀取所有 FASTA 格式的序列,包含標頭和序列內容
- 序列豐度提取:從每個序列標頭提取豐度值
- TOP K 序列選取:根據 top_k 參數選取豐度最高的 K 條序列
- 單檔案限制:每個樣本只處理一個檔案,若發現多個檔案會記錄警告並使用第一個
- 豐度解析容錯:標頭豐度解析失敗時使用 0.0 作為預設值
- 完整序列保留:每條選中的序列都會完整保留,不進行截斷或修改
採用與 path.md / locus_qc.md 對齊的欄位策略:
>batch={batch_id};locus={locus};sample={sample_id};src={source};rule={rule};conf={confidence};origin={origin};asv={asv_id};abundance={abundance}
現狀:僅使用簡化形式 >sample|src=...|conf=...;未來升級時需保持向後相容(提供轉換腳本)。
| 面向 | Debug 模式 | Recommendation 模式 |
|---|---|---|
| 目的 | 全面追蹤 | 最終下載/分析 |
| 檔案量 | 大 | 小 |
| 目錄 | extracted_sequences/ | qcResult/ (recommended_sequences.fas) |
| Header | step_id + 原標頭 | 統一結構化 |
| 再處理需求 | 開發/驗證 | 下游分析/回顧 |
所有萃取的序列都會重新標記標頭,遵循統一的命名規範:
>{step_id}|{原始標頭}
標頭命名範例:
>denoise_r1|{原標頭}
>merger_r1|{原標頭}
例如:
>merger_r1|Abrodictyum_cumingii_ZXC002739_KTHU1461_01_0.551_abundance_59_r1
每個 sample_id 產生一個 {sample_id}.fas 檔案,內容為該樣本於所有指定來源目錄下的所有序列,並依來源步驟標註標頭。
{locus}_result/qcResult/extracted_sequences/
├── {sample_id}.fas
├── ...
└── {sample_id}.fas
假設樣本 Abrodictyum_cumingii_ZXC002739_KTHU1461 的完整輸出:
# 檔案:extracted_sequences/Abrodictyum_cumingii_ZXC002739_KTHU1461.fas
>denoise_r1|Abrodictyum_cumingii_ZXC002739_KTHU1461_01_r1_0.750_abundance_150
ATCGATCGATCGATCGATCGATCG...
>denoise_r2|Abrodictyum_cumingii_ZXC002739_KTHU1461_01_r2_0.720_abundance_140
GCTAGCTAGCTAGCTAGCTAGCTA...
>merger_r1|Abrodictyum_cumingii_ZXC002739_KTHU1461_01_0.551_abundance_59_r1
ATCGATCGATCGATCGATCGATCG...
>merger_merged|Abrodictyum_cumingii_ZXC002739_KTHU1461_01_0.551_abundance_59
ATCGATCGATCGNNNNNGCTAGCTAGCTA...
>merger_r1ref|Abrodictyum_cumingii_ZXC002739_KTHU1461_01_0.551_abundance_59_r1
ATCGATCGATCGATCGATCGATCG...
>merger_r1ref|reference_sequence_for_Abrodictyum_cumingii
ATCGATCGATCGATCGATCGATCGAAAA...
# 主要介面
def combine_sample_sequences(sample_id: str, top_k: int = 3) -> str
def combine_all_samples(sample_id_list: List[str], top_k: int = 3) -> Dict[str, str]
# 檔案級別 TOP K 邏輯(多檔單序列步驟)
def _extract_sequences_from_multiple_files(source_dir: str, sample_id: str, step_id: str, top_k: int = 3) -> List[Tuple[str, str]]
def _extract_abundance_from_filename(filename: str) -> float
def _read_all_sequences_from_file(file_path: str, step_id: str) -> List[Tuple[str, str]]
# 序列級別 TOP K 邏輯(一檔多序列步驟)
def _extract_sequences_from_single_file(source_dir: str, sample_id: str, step_id: str) -> List[Tuple[str, str]]
def _extract_top_k_sequences(sequences: List[Tuple[str, str]], top_k: int = 3) -> List[Tuple[str, str]]
def _extract_abundance_percentage(header: str) -> float單樣本序列萃取:
combine_sample_sequences() 函數負責萃取單一樣本的序列並合併為 FASTA 格式。接受樣本識別碼和 TOP K 參數,返回合併後的 FASTA 內容。當找不到任何來源檔案時會拋出 FileNotFoundError。
批量樣本序列萃取:
combine_all_samples() 函數提供批量處理功能,接受樣本識別碼列表,返回每個樣本對應的 FASTA 內容字典。適用於需要同時處理多個樣本的場景。
| 參數 | 型別 | 預設值 | 說明 |
|---|---|---|---|
top_k |
int |
3 |
TOP K 值:檔案數(多檔單序列)或序列數(一檔多序列) |
sample_id |
str |
- | 樣本識別碼 |
step_id |
str |
- | 步驟標識(如 merger_r1ref) |
source_dir |
str |
- | 來源目錄路徑 |
sequences |
List[Tuple[str, str]] |
- | 序列列表 (header, sequence) |
top_k = None或top_k = 0:保留全部檔案/序列(向前相容)top_k < 0:視為 0,保留全部檔案/序列- 檔案數/序列數 ≤
top_k:保留全部檔案/序列 - 豐度解析失敗:豐度值設為
0.0
| 文件 | 對齊內容 |
|---|---|
locus_qc.md |
recommendation_source, confidence clamp |
path.md |
FASTA header 統一欄位草案 |
llm.md |
fallback_reason & provenance 來源欄位 |
log_flow.md |
非致命錯誤日誌分類 |
