import subprocess
from pathlib import Path
from latch import small_task, workflow
from latch.types import LatchFile
@small_task
def assembly_task(read1: LatchFile, read2: LatchFile) -> LatchFile:
# A reference to our output.
sam_file = Path("covid_assembly.sam").resolve()
_bowtie2_cmd = [
"bowtie2/bowtie2",
"--local",
"-x",
"wuhan",
"-1",
read1.local_path,
"-2",
read2.local_path,
"--very-sensitive-local",
"-S",
str(sam_file),
]
subprocess.run(_bowtie2_cmd)
return LatchFile(str(sam_file), "latch:///covid_assembly.sam")
@small_task
def sort_bam_task(sam: LatchFile) -> LatchFile:
bam_file = Path("covid_sorted.bam").resolve()
_samtools_sort_cmd = [
"samtools",
"sort",
"-o",
str(bam_file),
"-O",
"bam",
sam.local_path,
]
subprocess.run(_samtools_sort_cmd)
return LatchFile(str(bam_file), "latch:///covid_sorted.bam")
@workflow
def assemble_and_sort(read1: LatchFile, read2: LatchFile) -> LatchFile:
"""Description...
markdown header
----
Write some documentation about your workflow in
markdown here:
> Regular markdown constructs work as expected.
# Heading
* content1
* content2
__metadata__:
display_name: Assemble and Sort FastQ Files
author:
name:
email:
github:
repository:
license:
id: MIT
Args:
read1:
Paired-end read 1 file to be assembled.
__metadata__:
display_name: Read1
read2:
Paired-end read 2 file to be assembled.
__metadata__:
display_name: Read2
"""
sam = assembly_task(read1=read1, read2=read2)
return sort_bam_task(sam=sam)