Merge pull request #812 from rhpvorderman/multithread_bam_read

Multithread bam read

Author: marcelm

CHANGES.rst

@@ -16,6 +16,8 @@ development version
   Option ``-Z`` (equivalent to ``--compression-level=1``) is now deprecated.
 * The previously hidden option ``--compression-level`` is now shown in the
   ``--help`` output.
+* :issue:`812`: Fixed an issue where multithreaded reading of unaligned BAM
+  files would cause an error.
 * :issue:`820`: On Bioconda, Cutadapt is now also available for ARM64 Macs (M1/M2).
 * :issue:`817`: Fixed missing statistics when ``--revcomp`` was used for
   paired-end data.

pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
 requires-python = ">=3.9"
 dynamic = ["version"]
 dependencies = [
-    "dnaio >= 1.2.0",
+    "dnaio >= 1.2.3",
     "xopen >= 1.6.0",
 ]
 

src/cutadapt/files.py

@@ -95,14 +95,21 @@ def __init__(
         self,
         *files: BinaryIO,
         interleaved: bool = False,
+        fileformat=None,
     ):
         self._files = files
         self.interleaved = interleaved
+        self.fileformat = fileformat
         for f in self._files:
             assert f is not None
 
     def open(self):
-        return dnaio.open(*self._files, interleaved=self.interleaved, mode="r")
+        return dnaio.open(
+            *self._files,
+            interleaved=self.interleaved,
+            mode="r",
+            fileformat=self.fileformat,
+        )
 
     def close(self) -> None:
         for file in self._files:

src/cutadapt/runners.py

@@ -158,6 +158,7 @@ def __init__(
         read_pipe: Connection,
         write_pipe: Connection,
         need_work_queue: multiprocessing.Queue,
+        file_format,
     ):
         super().__init__()
         self._id = id_
@@ -168,6 +169,7 @@ def __init__(
         self._write_pipe = write_pipe
         self._need_work_queue = need_work_queue
         self._proxy_files = proxy_files
+        self._file_format = file_format
 
     def run(self):
         try:
@@ -189,7 +191,11 @@ def run(self):
                     io.BytesIO(self._read_pipe.recv_bytes())
                     for _ in range(self._n_input_files)
                 ]
-                infiles = InputFiles(*files, interleaved=self._interleaved_input)
+                infiles = InputFiles(
+                    *files,
+                    interleaved=self._interleaved_input,
+                    fileformat=self._file_format,
+                )
                 (n, bp1, bp2) = self._pipeline.process_reads(infiles)
                 stats += Statistics().collect(n, bp1, bp2, [], [])
                 self._send_outfiles(chunk_index, n)
@@ -320,7 +326,13 @@ def __init__(
         )
         self._reader_process.daemon = True
         self._reader_process.start()
-        self._input_file_format = self._try_receive(file_format_connection_r)
+        self._input_file_format: FileFormat = self._try_receive(
+            file_format_connection_r
+        )
+        self._file_format_string = self._input_file_format.name.lower()
+        if self._file_format_string == "bam":
+            # Individual BAM record chunks will have no header
+            self._file_format_string = "bam_no_header"
 
     def _start_workers(
         self, pipeline, proxy_files
@@ -338,6 +350,7 @@ def _start_workers(
                 self._connections[index],
                 conn_w,
                 self._need_work_queue,
+                file_format=self._file_format_string,
             )
             worker.daemon = True
             worker.start()

tests/cut/small_from_bam.fastq

@@ -0,0 +1,12 @@
+@prefix:1_13_573
+CGTCCGAANTAGCTACCACCCTGA
++
+)3%)&&&&!.1&(6:<'67..*,:
+@prefix:1_13_1259
+AGCCGCTANGACGGGTTGGCCC
++
+;<:&:A;A!9<<<,7:<=3=;:
+@prefix:1_13_1440
+CAAGATCTNCCCTGCCACATTGCCCTAGTTAAAC
++
+<=A:A=57!7<';<6?5;;6:+:=)71>70<,=:

tests/data/small.bam

@@ -79,6 +79,14 @@ def test_small(run):
     run("-a TTAGACATATCTCCGTCG", "small.fastq", "small.fastq")
 
 
+def test_small_bam(run, cores):
+    run(
+        "--cores {} -a TTAGACATATCTCCGTCG".format(cores),
+        "small_from_bam.fastq",
+        "small.bam",
+    )
+
+
 def test_empty_fastq(run, cores):
     run("--cores {} -a TTAGACATATCTCCGTCG".format(cores), "empty.fastq", "empty.fastq")