-
Notifications
You must be signed in to change notification settings - Fork 34
/
bowtie_index.py
77 lines (61 loc) · 2.36 KB
/
bowtie_index.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
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
"""Create genome index for Bowtie aligner."""
import shutil
from pathlib import Path
from plumbum import TEE
from resolwe.process import Cmd, DataField, DirField, FileField, Process, StringField
class BowtieIndex(Process):
"""Create Bowtie genome index."""
slug = "bowtie-index"
process_type = "data:index:bowtie"
name = "Bowtie genome index"
requirements = {
"expression-engine": "jinja",
"executor": {
"docker": {"image": "public.ecr.aws/genialis/resolwebio/rnaseq:6.0.0"},
},
"resources": {
"cores": 4,
"memory": 16384,
},
}
category = "Genome index"
data_name = '{{ ref_seq.fasta.file|basename|default("?") }}'
version = "1.2.1"
class Input:
"""Input fields for BowtieIndex."""
ref_seq = DataField(
"seq:nucleotide", label="Reference sequence (nucleotide FASTA)"
)
class Output:
"""Output fields to process BowtieIndex."""
index = DirField(label="Bowtie index")
fastagz = FileField(label="FASTA file (compressed)")
fasta = FileField(label="FASTA file")
fai = FileField(label="FASTA file index")
species = StringField(label="Species")
build = StringField(label="Build")
def run(self, inputs, outputs):
"""Run analysis."""
basename = Path(inputs.ref_seq.output.fasta.path).name
assert basename.endswith(".fasta")
name = basename[:-6]
index_dir = Path("bowtie_index")
index_dir.mkdir()
shutil.copy(Path(inputs.ref_seq.output.fasta.path), Path.cwd())
shutil.copy(Path(inputs.ref_seq.output.fastagz.path), Path.cwd())
shutil.copy(Path(inputs.ref_seq.output.fai.path), Path.cwd())
args = [
inputs.ref_seq.output.fasta.path,
index_dir / f"{name}_index",
"--threads",
self.requirements.resources.cores,
]
return_code, _, _ = Cmd["bowtie-build"][args] & TEE(retcode=None)
if return_code:
self.error("Error occurred while preparing the Bowtie index.")
outputs.index = index_dir.name
outputs.fasta = f"{name}.fasta"
outputs.fastagz = f"{name}.fasta.gz"
outputs.fai = f"{name}.fasta.fai"
outputs.species = inputs.ref_seq.output.species
outputs.build = inputs.ref_seq.output.build