Skip to main contentscientific-metagenome-assembled-genomes
メタゲノムアセンブルゲノム (MAG) 解析スキル。
MetaBAT2 / CONCOCT / MaxBin2 ビニング・CheckM2 品質評価・
GTDB-Tk 分類学的分類・dRep 脱重複・Prokka アノテーション・
MAG アセンブリ品質レポートパイプライン。
TU 外スキル (CLI ラッパー + Python ライブラリ)。
설치로 이동 Codex 또는 Claude로 설치 이 Prompt를 복사해 Codex, Claude 또는 다른 어시스턴트에 붙여 넣으면 Skill 페이지를 검토하고 설치를 진행할 수 있습니다.
직접 명령은 검토 Prompt를 거치지 않습니다. 실행하기 전에 소스를 확인하세요.
npx skills add https://github.com/nahisaho/satori --skill scientific-metagenome-assembled-genomes명령은 한 줄로 유지됩니다. 복사하기 전에 가로로 스크롤해 전체 내용을 확인하세요.
로컬 사본을 원하시나요? SkillsMP에서 현재 제공할 수 있는 파일을 다운로드하세요.
| name | scientific-metagenome-assembled-genomes |
| description | メタゲノムアセンブルゲノム (MAG) 解析スキル。
MetaBAT2 / CONCOCT / MaxBin2 ビニング・CheckM2 品質評価・
GTDB-Tk 分類学的分類・dRep 脱重複・Prokka アノテーション・
MAG アセンブリ品質レポートパイプライン。
TU 外スキル (CLI ラッパー + Python ライブラリ)。
|
| tu_tools | [{"key":"mgnify","name":"MGnify","description":"メタゲノムアセンブル・MAG データ検索"}] |
Scientific Metagenome-Assembled Genomes
メタゲノムリードから個別ゲノム (MAG) を再構築する
ビニング・品質評価・分類・アノテーションの
統合パイプラインを提供する。
When to Use
- メタゲノムショットガンデータから MAG を再構築するとき
- コンティグビニング (MetaBAT2/CONCOCT/MaxBin2) を実行するとき
- CheckM/CheckM2 でゲノム完全性・コンタミネーションを評価するとき
- GTDB-Tk で MAG の分類学的位置づけを行うとき
- dRep で冗長な MAG を脱重複するとき
- Prokka/Bakta で MAG のアノテーションを行うとき
Quick Start
1. MetaBAT2 ビニング
import subprocess
import pandas as pd
from pathlib import Path
def run_metabat2(assembly_fasta, bam_file,
output_dir="metabat2_bins",
min_contig=2500):
"""
MetaBAT2 — メタゲノムコンティグビニング。
Parameters:
assembly_fasta: str — アセンブリ FASTA
bam_file: str — ソート済み BAM
output_dir: str — 出力ディレクトリ
min_contig: int — 最小コンティグ長
"""
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
depth_file = out /
subprocess.run([
,
, (depth_file),
bam_file
], check=)
subprocess.run([
,
, assembly_fasta,
, (depth_file),
, (out / ),
, (min_contig),
, ,
], check=)
bins = (out.glob())
()
bins
"depth.txt"
"jgi_summarize_bam_contig_depths"
"--outputDepth"
str
True
"metabat2"
"-i"
"-a"
str
"-o"
str
"bin"
"-m"
str
"--seed"
"42"
True
list
"bin.*.fa"
print
f"MetaBAT2: {len(bins)} bins generated"
return
2. CheckM2 品質評価
def run_checkm2(bin_dir, output_dir="checkm2_out",
threads=8):
"""
CheckM2 — MAG 品質評価
(完全性 / コンタミネーション / N50)。
Parameters:
bin_dir: str — ビンディレクトリ
output_dir: str — 出力ディレクトリ
threads: int — スレッド数
"""
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
subprocess.run([
"checkm2", "predict",
"--input", bin_dir,
"--output-directory", str(out),
"--threads", str(threads),
"-x", "fa",
], check=True)
report = out / "quality_report.tsv"
df = pd.read_csv(report, sep="\t")
df["quality"] = df.apply(
lambda r: (
"high" if r["Completeness"] >= 90
and r["Contamination"] < 5
else "medium"
if r["Completeness"] >= 50
and r["Contamination"] < 10
else "low"), axis=1)
n_hq = (df["quality"] == "high").sum()
n_mq = (df["quality"] == "medium").sum()
n_lq = (df["quality"] == "low").sum()
print(f"CheckM2: {n_hq} HQ, {n_mq} MQ, "
f"{n_lq} LQ MAGs")
return df
def filter_quality_mags(checkm_df,
min_completeness=50,
max_contamination=10):
"""
品質基準によるMAGフィルタリング。
Parameters:
checkm_df: pd.DataFrame — CheckM2 結果
min_completeness: float — 最小完全性 (%)
max_contamination: float — 最大汚染 (%)
"""
filtered = checkm_df[
(checkm_df["Completeness"]
>= min_completeness)
& (checkm_df["Contamination"]
<= max_contamination)
].copy()
print(f"Filter: {len(filtered)}/"
f"{len(checkm_df)} MAGs passed "
f"(≥{min_completeness}% comp, "
f"≤{max_contamination}% contam)")
return filtered
3. GTDB-Tk 分類
def run_gtdbtk(bin_dir, output_dir="gtdbtk_out",
threads=8):
"""
GTDB-Tk — ゲノム分類学分類
(GTDB taxonomy)。
Parameters:
bin_dir: str — フィルタ済みビンディレクトリ
output_dir: str — 出力ディレクトリ
threads: int — スレッド数
"""
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
subprocess.run([
"gtdbtk", "classify_wf",
"--genome_dir", bin_dir,
"--out_dir", str(out),
"--cpus", str(threads),
"-x", "fa",
], check=True)
results = []
for domain in ["bac120", "ar53"]:
tsv = (out / f"gtdbtk.{domain}."
"summary.tsv")
if tsv.exists():
df = pd.read_csv(tsv, sep="\t")
df["domain_marker"] = domain
results.append(df)
if results:
combined = pd.concat(results,
ignore_index=True)
print(f"GTDB-Tk: {len(combined)} MAGs "
f"classified")
return combined
print("GTDB-Tk: no classification results")
return pd.DataFrame()
4. dRep 脱重複
def run_drep(bin_dir, output_dir="drep_out",
ani_threshold=0.95):
"""
dRep — MAG 脱重複 (ANI ベース)。
Parameters:
bin_dir: str — ビンディレクトリ
output_dir: str — 出力ディレクトリ
ani_threshold: float — ANI 閾値
"""
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
subprocess.run([
"dRep", "dereplicate",
str(out),
"-g", f"{bin_dir}/*.fa",
"-sa", str(ani_threshold),
"--ignoreGenomeQuality",
], check=True)
derep = list(
(out / "dereplicated_genomes").glob("*.fa"))
print(f"dRep: {len(derep)} dereplicated MAGs "
f"(ANI ≥ {ani_threshold})")
return derep
5. MAG パイプライン統合
def mag_pipeline(assembly_fasta, bam_file,
output_dir="mag_results",
threads=8):
"""
MAG 統合パイプライン。
Parameters:
assembly_fasta: str — メタゲノムアセンブリ
bam_file: str — ソート済み BAM
output_dir: str — 出力ルート
threads: int — スレッド数
"""
out = Path(output_dir)
out.mkdir(parents=True, exist_ok=True)
bins = run_metabat2(
assembly_fasta, bam_file,
str(out / "bins"))
checkm = run_checkm2(
str(out / "bins"),
str(out / "checkm2"),
threads)
quality = filter_quality_mags(checkm)
taxonomy = run_gtdbtk(
str(out / "bins"),
str(out / "gtdbtk"),
threads)
derep = run_drep(
str(out / "bins"),
str(out / "drep"))
print(f"MAG pipeline: {len(bins)} bins → "
f"{len(quality)} QC passed → "
f"{len(derep)} dereplicated")
results = Path(output_dir) / "results"
results.mkdir(parents=True, exist_ok=True)
quality.to_csv(results / "mag_quality_summary.csv", index=False)
print(f" ✔ MAG quality summary: {results / 'mag_quality_summary.csv'}")
if not taxonomy.empty:
taxonomy.to_csv(results / "mag_taxonomy.csv", index=False)
print(f" ✔ MAG taxonomy: {results / 'mag_taxonomy.csv'}")
representative_fasta = results / "representative_mags.fasta"
with open(representative_fasta, "w") as f:
for mag_path in derep:
mag_name = Path(mag_path).stem
with open(mag_path) as mag_f:
for line in mag_f:
if line.startswith(">"):
f.write(f">{mag_name}_{line[1:]}")
else:
f.write(line)
print(f" ✔ Representative MAGs FASTA: {representative_fasta}")
import json
pipeline_summary = {
"total_bins": len(bins),
"quality_passed": len(quality),
"high_quality": int((quality["quality"] == "high").sum()) if "quality" in quality.columns else 0,
"medium_quality": int((quality["quality"] == "medium").sum()) if "quality" in quality.columns else 0,
"dereplicated": len(derep),
"classified": len(taxonomy) if not taxonomy.empty else 0,
}
with open(results / "mag_pipeline_summary.json", "w") as f:
json.dump(pipeline_summary, f, indent=2)
print(f" ✔ Pipeline summary: {results / 'mag_pipeline_summary.json'}")
return {
"bins": bins,
"checkm": checkm,
"quality": quality,
"taxonomy": taxonomy,
"dereplicated": derep,
}
パイプライン統合
microbiome-metagenomics → metagenome-assembled-genomes → environmental-ecology
(メタゲノム組成解析) (MAG 再構築) (生態系統合)
│ │ ↓
long-read-sequencing ─────────┘ phylogenomics
(ロングリードアセンブリ) (系統解析)
パイプライン出力
| ファイル | 説明 | 次スキル |
|---|
*_bins/bin.*.fa | ビンゲノム | → dRep, GTDB-Tk |
checkm2_out/quality_report.tsv | 品質レポート | → フィルタリング |
gtdbtk_out/*.summary.tsv | 分類結果 | → phylogenomics |
drep_out/dereplicated_genomes/ | 脱重複 MAG | → environmental-ecology |
ToolUniverse 連携
| TU Key | ツール名 | 連携内容 |
|---|
mgnify | MGnify | メタゲノムアセンブル・MAG データ検索 |