85 lines
2.4 KiB
Python
Executable File
85 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
RNA-seq Quantification Skill Runner
|
|
|
|
Executes the RNA-seq quantification stage as a skill.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, Any
|
|
|
|
|
|
def run_skill(
|
|
input_dir: Path,
|
|
gtf: Path,
|
|
output_dir: Path,
|
|
prefix: str = "",
|
|
cpu: int = 4,
|
|
sample_config: Path = None,
|
|
job_id: str = "unknown"
|
|
) -> Dict[str, Any]:
|
|
"""Run the RNA-seq quantification skill"""
|
|
|
|
# Import the main module
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "apps" / "cli"))
|
|
from run_rnaseq_quant import main as quant_main
|
|
|
|
# Build arguments
|
|
args = [
|
|
"-i", str(input_dir),
|
|
"-g", str(gtf),
|
|
"-o", str(output_dir),
|
|
"-j", job_id,
|
|
"-p", prefix,
|
|
"-t", str(cpu)
|
|
]
|
|
|
|
if sample_config:
|
|
args.extend(["--sample-config", str(sample_config)])
|
|
|
|
# Run the quantification
|
|
sys.argv = ["run_rnaseq_quant.py"] + args
|
|
quant_main()
|
|
|
|
# Load and return manifest
|
|
manifest_path = output_dir / "final" / f"{prefix}.rnaseq_quant.manifest.json"
|
|
if manifest_path.exists():
|
|
with open(manifest_path, 'r') as f:
|
|
return json.load(f)
|
|
|
|
return {"status": "failed", "error": "Manifest not found"}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="RNA-seq Quantification Skill")
|
|
parser.add_argument("-i", "--input", required=True, help="Input directory with BAM files")
|
|
parser.add_argument("-g", "--gtf", required=True, help="Reference genome GTF")
|
|
parser.add_argument("-o", "--output", required=True, help="Output directory")
|
|
parser.add_argument("-j", "--job-id", default="unknown", help="Job ID")
|
|
parser.add_argument("-p", "--prefix", default="", help="Output prefix")
|
|
parser.add_argument("-t", "--cpu", type=int, default=4, help="CPU threads")
|
|
parser.add_argument("-c", "--config", help="Sample config YAML")
|
|
|
|
args = parser.parse_args()
|
|
|
|
result = run_skill(
|
|
input_dir=Path(args.input),
|
|
gtf=Path(args.gtf),
|
|
output_dir=Path(args.output),
|
|
prefix=args.prefix,
|
|
cpu=args.cpu,
|
|
sample_config=Path(args.config) if args.config else None,
|
|
job_id=args.job_id
|
|
)
|
|
|
|
print(json.dumps(result, indent=2))
|
|
return 0 if result.get("status") == "success" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|