aligner/index.ts

import { spawn } from 'child_process';
import { cpus } from 'os';
import fs from 'fs'
import path from 'path';

const async_exec = (prog: string, args: string[], onData: Function) => {
    return new Promise((resolve, reject) => {
        const child = spawn(prog, args, {shell: true})

        child.stdout.on('data', data => onData(data.toString().trim()))
        child.stderr.on('data', data => onData(data.toString().trim()))

        child.on('error', err => reject(err))
        child.on('exit', code => resolve(code))
    })
}

const asyncBwaMem = (
    refPath   : string,
    reads     : string | Array<string> | Array<Array<string>>,
    // R1        : string | Array<string>,
    // R2        : string | Array<string>,
    runName   : string,
    libName   : string,
    outputDir : string,
    onData    : Function,
    options?  : any,
    ) => {

    const defaultOptions = {
      output_discordant: true,
      output_splitted: true,
      output_unmapped: true
    }

    if (typeof options === 'undefined') {
      options = defaultOptions
    } else {
      options = {...defaultOptions, ...options}
    }

    const refName = path.parse(refPath).name
  
    const bwa        = 'bwa'
    const samblaster = 'samblaster'
    const samtools   = 'samtools'
    const sambamba   = 'sambamba'

    let readsIn: string
    let isPairedEnd = false

    if (Array.isArray(reads) ) {
      isPairedEnd = true
      console.log('Assuming paired end reads');
      
      const [R1, R2] = reads
      const R1_arr = Array.isArray(R1) ? R1.join(' ') : R1
      const R2_arr = Array.isArray(R2) ? R2.join(' ') : R2
    
      const R1_kitty = R1_arr.slice(-2) === 'gz' ? 'zcat' : 'cat'
      const R2_kitty = R2_arr.slice(-2) === 'gz' ? 'zcat' : 'cat'
    
      const R1_in = `'< ${R1_kitty} ${R1_arr}'`
      const R2_in = `'< ${R2_kitty} ${R2_arr}'`
      readsIn = R1_in + ' ' + R2_in
    } else {
      readsIn = reads
    }

    let bam       = path.join(outputDir, `bwa_mem_properly_on_${refName}.bam`)
    let bamSorted = path.join(outputDir, `bwa_mem_properly_on_${refName}.sorted.bam`)
    let retObj: any = { bamSorted } 
    if(options?.remove_mapped) {
      bam = '/dev/null'
      delete retObj.bamSorted
    }

    const threads = String(cpus().length)

    let samblasterCmd: Array<string> = []
    if (options?.output_discordant || options?.output_splitted) {
      console.log('Using samblaster');

      // https://github.com/GregoryFaust/samblaster
      samblasterCmd = ['|', samblaster, 
        '--addMateTags',
        '-a', // Accept duplicate marks already in input file
        '-e', // Exclude reads marked as duplicates from discordant, splitter, and/or unmapped
      ]

      if(options?.output_discordant) {
        if(!isPairedEnd) {
          console.log('Discordant reads can be found only in paired reads, skipping')
        } else {
          const discordantFile = path.join(outputDir, `bwa_mem_discordants_on_${refName}.sam`)
          console.log('Discordant reads file path: ', discordantFile);
          samblasterCmd = [...samblasterCmd, '-d', discordantFile]
          retObj = {...retObj, discordantFile}
        }
      }

      if (!isPairedEnd) {
        samblasterCmd = [...samblasterCmd, '--ignoreUnmated']
      }

      if(options?.output_splitted) {
        const splitterFile   = path.join(outputDir, `bwa_mem_splitters_on_${refName}.sam`)
        console.log('Splitted reads file path: ', splitterFile);
        samblasterCmd = [...samblasterCmd, '-s', splitterFile]
        retObj = {...retObj, splitterFile}
      }

      if(options?.output_unmapped) {
        const unmappedFile   = path.join(outputDir, `bwa_mem_unmapped_on_${refName}.fq`)
        console.log('Unmapped reads file path: ', unmappedFile);
        samblasterCmd = [...samblasterCmd, '-u', unmappedFile]
        retObj = {...retObj, unmappedFile}
      }
    }
  
    return new Promise<string[]>(async (resolve, reject) => {
      try {
        const code = await async_exec(
        bwa, ['mem',
          '-t', threads,
          '-R', `"@RG\\tPL:Illumina\\tID:${+(new Date)}\\tSM:${runName}\\tLB:${libName}"`,
          refPath, 
          readsIn,
        ...samblasterCmd,
        '|',
        samtools,
          'view',
          '-Sb',
          '-',
        '>',
        bam
        ], (message: string) => onData('[BWA-MEM] ' + message))
        onData('[BWA-MEM][EXIT CODE] ' + code)
        
        if(retObj.bamSorted) {
          const code_sort = await async_exec(
            sambamba, ['sort',
              '-t', threads,
              bam
          ], (message: string) => onData('[SAMBAMBA-SORT] ' + message))
          onData('[SAMBAMBA-SORT][EXIT CODE] ' + code_sort)
          fs.unlinkSync(bam)
        }
        
        resolve(retObj)
      } catch (err) {
        reject(err)
      }
    })
}

export { asyncBwaMem }