added filtering in merge and sort tools (see #81)

Author: lomereiter

sambamba/merge.d

@@ -92,6 +92,7 @@ import core.atomic;
 
 import sambamba.utils.common.progressbar;
 import sambamba.utils.common.overwrite;
+import sambamba.utils.common.filtering;
 
 void printUsage() {
     stderr.writeln("Usage: sambamba-merge [options] <output.bam> <input1.bam> <input2.bam> [...]");
@@ -104,6 +105,8 @@ void printUsage() {
     stderr.writeln("               output merged header to stdout in SAM format, other options are ignored; mainly for debug purposes");
     stderr.writeln("         -p, --show-progress");
     stderr.writeln("               show progress bar in STDERR");
+    stderr.writeln("         -F, --filter=FILTER");
+    stderr.writeln("               keep only reads that satisfy FILTER");
 }
 
 // these variables can be implicitly used in tasks created in writeBAM
@@ -114,6 +117,7 @@ shared(string[string][]) program_id_map;
 shared(string[string][]) readgroup_id_map;
 
 __gshared static TaskPool task_pool;
+__gshared static Filter read_filter;
 
 BamRead changeAlignment(Tuple!(BamRead, size_t) al_with_file_id) {
     auto al = al_with_file_id[0];
@@ -157,14 +161,10 @@ BamRead changeAlignment(Tuple!(BamRead, size_t) al_with_file_id) {
 }
 
 auto modifyAlignmentRange(T)(T alignments_with_file_id) {
-    version(serial) {
-        return map!changeAlignment(zip(alignments_with_file_id[0], 
-                                       repeat(alignments_with_file_id[1])));
-    } else {
-        return task_pool.map!changeAlignment(zip(alignments_with_file_id[0],
-                                                 repeat(alignments_with_file_id[1])),
-                                            8192);
-    }
+  return task_pool.map!changeAlignment(
+          zip(filtered(alignments_with_file_id[0], read_filter),
+              repeat(alignments_with_file_id[1])),
+          8192);
 }
 
 version(standalone) {
@@ -180,6 +180,7 @@ int merge_main(string[] args) {
     bool validate_headers = false;
     bool header_only = false;
     bool show_progress = false;
+    string filter_str = null;
 
     if (args.length < 4) {
         printUsage();
@@ -194,7 +195,10 @@ int merge_main(string[] args) {
                "compression-level|l",   &compression_level,
                "validate-headers|v",    &validate_headers,
                "header|H",              &header_only,
-               "show-progress|p",       &show_progress);
+               "show-progress|p",       &show_progress,
+               "filter|F",              &filter_str);
+
+        read_filter = createFilterFromQuery(filter_str);
 
         task_pool = new TaskPool(number_of_threads);
         scope(exit) task_pool.finish();

sambamba/sort.d

@@ -47,6 +47,7 @@ import std.c.string;
 import sambamba.utils.common.progressbar;
 import sambamba.utils.common.overwrite;
 import sambamba.utils.common.tmpdir;
+import sambamba.utils.common.filtering;
 
 import thirdparty.mergesort;
 
@@ -69,6 +70,8 @@ void printUsage() {
     stderr.writeln("               show progressbar in STDERR");
     stderr.writeln("         -t, --nthreads=NTHREADS");
     stderr.writeln("               use specified number of threads");
+    stderr.writeln("         -F, --filter=FILTER");
+    stderr.writeln("               keep only reads that satisfy FILTER");
 }
 
 version(standalone) {
@@ -94,6 +97,7 @@ class Sorter {
     bool uncompressed_chunks = false;
     string output_filename = null;
     string filename = null;
+    string filter_str = null;
 
     struct UnsortedChunk {
         size_t max_sz;
@@ -181,20 +185,25 @@ class Sorter {
     }
 
     void sort() {
+        auto filter = createFilterFromQuery(filter_str);
+
         createHeader();
 
         bam.setBufferSize(16_000_000);
         bam.assumeSequentialProcessing();
+
         if (show_progress) {
             stderr.writeln("Writing sorted chunks to temporary directory...");
             bar = new shared(ProgressBar)();
             auto reads = bam.readsWithProgress(
                                   (lazy float p){ bar.update(p); }
                               );
-            writeSortedChunks(reads);
+            auto filtered_reads = filtered(reads, filter);
+            writeSortedChunks(filtered_reads);
             bar.finish();
         } else {
-            writeSortedChunks(bam.reads!withoutOffsets);
+            auto filtered_reads = filtered(bam.reads!withoutOffsets(), filter);
+            writeSortedChunks(filtered_reads);
         }
 
         scope(success) {
@@ -427,7 +436,8 @@ int sort_main(string[] args) {
                "uncompressed-chunks|u", &sorter.uncompressed_chunks,
                "compression-level|l",   &sorter.compression_level,
                "show-progress|p",       &show_progress,
-               "nthreads|t",            &n_threads);
+               "nthreads|t",            &n_threads,
+               "filter|F",              &sorter.filter_str);
 
         if (sorter.output_filename is null) {
             sorter.output_filename = setExtension(args[1], "sorted.bam");