-
Notifications
You must be signed in to change notification settings - Fork 175
/
wrapper.py
48 lines (35 loc) · 1.19 KB
/
wrapper.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
__author__ = "Filipe G. Vieira"
__copyright__ = "Copyright 2021, Filipe G. Vieira"
__license__ = "MIT"
import os
import random
import tempfile
from pathlib import Path
from snakemake.shell import shell
log = snakemake.log_fmt_shell(stdout=False, stderr=True, append=True)
extra = snakemake.params.get("extra", "")
# File formats
in_name, in_format = os.path.splitext(snakemake.input[0])
in_format = in_format.lstrip(".")
out_name, out_format = os.path.splitext(snakemake.output[0])
out_format = out_format.lstrip(".")
index = snakemake.output.get("index", "")
if index:
index = f"indexfilename={index}"
metrics = snakemake.output.get("metrics", "")
if metrics:
metrics = f"M={metrics}"
with tempfile.TemporaryDirectory() as tmpdir:
# This folder must not exist; it is created by BamSorMaDup
tmpdir_bamsormadup = Path(tmpdir) / "bamsormadup_{:06d}".format(
random.randrange(10 ** 6)
)
shell(
"bamsormadup threads={snakemake.threads}"
" inputformat={in_format}"
" tmpfile={tmpdir_bamsormadup}"
" outputformat={out_format}"
" {index} {metrics} {extra}"
" < {snakemake.input[0]} > {snakemake.output[0]}"
" {log}"
)