Allow multithreaded BAM reading

Author: rhpvorderman

pyproject.toml

@@ -22,7 +22,7 @@ classifiers = [
 requires-python = ">=3.8"
 dynamic = ["version"]
 dependencies = [
-    "dnaio >= 1.2.0",
+    "dnaio >= 1.3.0",
     "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()