task rmdup_picard {
    String name
    String sorted_bam
    String output_dir
    String ref

    command <<<

        if [ ! -d ${output_dir}/fusion ];then
        mkdir ${output_dir}/fusion
        fi

        java -XX:+UseParallelGC -XX:ParallelGCThreads=2 -Xmx12G  -jar $PICARD MarkDuplicates \
        I=${sorted_bam} \
        O=${output_dir}/alignment/${name}.picard.rmdup.bam  \
        CREATE_INDEX=true \
        M=${output_dir}/alignment/${name}.picard.rmdup.metrics.txt  \
        R=${ref}

    >>>
    output {
        String rmdup_bam = "${output_dir}/alignment/${name}.picard.rmdup.bam"
    }

}

task fusion_calling {
    String name
    String rmdup_bam
    String output_dir
    String ref

    command <<<

        if [ ! -d ${output_dir}/fusion ];then
        mkdir ${output_dir}/fusion
        fi

        # Extract the discordant paired-end alignments.
        samtools view -b -F 1294 ${rmdup_bam} > ${output_dir}/fusion/${name}.discordants.bam

        # Extract the split-read alignments
        samtools view -h ${rmdup_bam} \
        | /dataseq/jmdna/software/lumpy-sv/scripts/extractSplitReads_BwaMem -i stdin \
        | samtools view -Sb - \
        > ${output_dir}/fusion/${name}.splitters.bam

        lumpyexpress \
        -B ${rmdup_bam} \
        -S ${output_dir}/fusion/${name}.splitters.bam \
        -D ${output_dir}/fusion/${name}.discordants.bam \
        -o ${output_dir}/fusion/${name}.fusion.vcf

        #        fusion.filter.pl ${output_dir}/fusion/${name}.fusion.vcf ${output_dir}/fusion/${name}.fusion.filter.vcf
        #
        #        svtyper \
        #        -B ${rmdup_bam} \
        #        -i ${output_dir}/fusion/${name}.fusion.filter.vcf \
        #        -T ${ref} \
        #        -o ${output_dir}/fusion/${name}.fusion.gt.vcf

    >>>

    output {
        String vcf = "${output_dir}/fusion/${name}.fusion.vcf"
    }

}

workflow call_fusion {

    String name
    String sorted_bam
    String rmdup_bam
    String output_dir
    String ref
    String umi

    if (umi) {
        call rmdup_picard {
            input:
                name=name,
                sorted_bam=sorted_bam,
                output_dir=output_dir,
                ref=ref
        }
        call fusion_calling {
            input:
                name=name,
                rmdup_bam=rmdup_picard.rmdup_bam,
                output_dir=output_dir,
                ref=ref
        }
    }

    if (!umi) {
        call fusion_calling {
            input:
                name=name,
                rmdup_bam=rmdup_bam,
                output_dir=output_dir,
                ref=ref
        }
    }

    output {
        String fusion_vcf = "${output_dir}/fusion/${name}.fusion.vcf"
    }
}