diff --git a/HISTORY.md b/HISTORY.md index e17dcbb..823914a 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -1,5 +1,23 @@ # History +1.8.5 (2026-09-07) +------------------ + +* Concatenates reads and contigs by block copy rather than round-tripping every record through BioPython. `concatenate_single_fastq` parsed both inputs into a list of `SeqRecord`s and wrote them out again, which costs roughly 5-10x the file size in RAM; on 250 MiB of short reads that was 27s and 1.8 GB of peak RSS, and is now 0.7s and 46 MB. Only the non-chromosomal short reads pass through here, so the saving is a transient GB or two rather than the whole run - but it landed immediately before Unicycler, which needs the memory itself. `Plass.get_depth_long` no longer builds its own list of records either. Thanks @[sanjaynagi-eit](https://github.com/sanjaynagi-eit) ([#90](https://github.com/gbouras13/plassembler/pull/90)) +* A wrong-format input is now caught by checking the first byte against the format's record marker (`@` or `>`) instead of by parsing every record, keeping that check at O(1) instead of O(file). One consequence: a truncated *uncompressed* FASTQ is no longer rejected, where full parsing used to raise. Truncated `.gz` input still fails, because the decompressor raises before the copy finishes +* Concatenation now writes to a `.tmp` sibling and renames it into place. A wrong-format second input, a truncated gzip or a full disk can no longer leave a half-written FASTQ where the run expects a complete one, and a failed re-run leaves any previous output intact +* An empty `chromosome.fasta` reaching `Assembly.combine_input_fastas` is now a fatal error with a message explaining it. `list(SeqIO.parse(...))[0]` used to raise `IndexError` there; streaming the records has no equivalent, and without an explicit check every depth and copy number would have been computed against a `combined.fasta` containing no chromosome + +1.8.4 (2026-08-18) +------------------ + +* Fixes the 75bp head and tail cropping of long reads, which had been silently ignored since `chopper` v0.11.0. From v0.11.0, `chopper` only applies `--headcrop`/`--tailcrop` when `--trim-approach fixed-crop` is also specified, so `plassembler` now passes this +* Bumps the minimum `chopper` version to v0.11.0 +* As a result, filtered long reads are 150bp shorter as originally intended, and reads falling below `--min_length` after cropping are now removed. Expect small changes to long read depths and plasmid copy number estimates compared to v1.8.3 +* Makes `chopper` failures fatal. Previously only the last process in the read filtering pipeline was checked, so a `chopper` that exited non-zero (for example, an old `chopper` rejecting `--trim-approach`) left a valid but empty `chopper_long_reads.fastq.gz` and the assembly continued with zero reads. `plassembler` now checks every stage, reports `chopper`'s own error message rather than only the path to the logfile, rejects empty filtered output, and exits. The per-stage process handling this builds on came from @[sanjaynagi-eit](https://github.com/sanjaynagi-eit) ([#88](https://github.com/gbouras13/plassembler/pull/88)), which also uncovered that `tests/test_data/end_to_end/input_half.fastq.gz` had always ended mid-record, so `chopper` had been failing on it unnoticed +* `plassembler` now warns if the installed `chopper` is older than v0.11.0, as version pins only bind when an environment is first created +* Compresses filtered long reads with `bgzip -@` rather than `gzip`, which is roughly 10x faster on a multi-core machine and produces slightly smaller output. BGZF is a valid gzip stream, so nothing downstream changes, and `gzip` is still used where `bgzip` is unavailable. Note that `bgzip` reaches plassembler only via `samtools`' dependency on `htslib`, which does not currently resolve on Apple Silicon - M-series users get the `gzip` fallback and no speedup. Thanks @[sanjaynagi-eit](https://github.com/sanjaynagi-eit) ([#88](https://github.com/gbouras13/plassembler/pull/88)) + 1.8.3 (2026-07-05) ------------------ diff --git a/README.md b/README.md index 944ebf2..50be2ce 100644 --- a/README.md +++ b/README.md @@ -232,7 +232,7 @@ You will then need to install the external dependencies separately, which can be * [Unicycler](https://github.com/rrwick/Unicycler) >=0.4.8 * [Minimap2](https://github.com/lh3/minimap2) >=2.11 * [fastp](https://github.com/OpenGene/fastp) >=0.24.2 -* [chopper](https://github.com/wdecoster/chopper) >=0.5.0 +* [chopper](https://github.com/wdecoster/chopper) >=0.11.0 * [mash](https://github.com/marbl/Mash) >=2.2 * [Raven](https://github.com/lbcb-sci/raven) >=1.8 * [Samtools](https://github.com/samtools/samtools) >=0.15.0 diff --git a/build/environment.yaml b/build/environment.yaml index 72814ae..aef95e9 100644 --- a/build/environment.yaml +++ b/build/environment.yaml @@ -8,7 +8,7 @@ dependencies: - unicycler >=0.4.8 - minimap2 >=2.11 - fastp >=0.24.2 - - chopper >=0.5.0 + - chopper >=0.11.0 - mash >=2.2 - raven-assembler >=1.8 - samtools >=0.15.0 diff --git a/docs/install.md b/docs/install.md index 40cd3b7..ca4a75c 100644 --- a/docs/install.md +++ b/docs/install.md @@ -32,7 +32,7 @@ You will then need to install the external dependencies separately, which can be * [Unicycler](https://github.com/rrwick/Unicycler) >=0.4.8 * [Minimap2](https://github.com/lh3/minimap2) >=2.11 * [fastp](https://github.com/OpenGene/fastp) >=0.24.2 -* [chopper](https://github.com/wdecoster/chopper) >=0.5.0 +* [chopper](https://github.com/wdecoster/chopper) >=0.11.0 * [mash](https://github.com/marbl/Mash) >=2.2 * [Raven](https://github.com/lbcb-sci/raven) >=1.8 * [Samtools](https://github.com/samtools/samtools) >=0.15.0 diff --git a/pyproject.toml b/pyproject.toml index 5084b4e..c8271c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "plassembler" -version = "1.8.3" +version = "1.8.5" description = "Quickly and accurately assemble plasmids in hybrid sequenced bacterial isolates" authors = [ { name = "George Bouras", email = "george.bouras@adelaide.edu.au" } @@ -75,7 +75,7 @@ flye = ">=2.9" unicycler = ">=0.4.8" minimap2 = ">=2.11" fastp = ">=0.24.2" -chopper = ">=0.5.0" +chopper = ">=0.11.0" mash = ">=2.2" raven-assembler = ">=1.8" samtools = ">=0.15.0" diff --git a/src/plassembler/__init__.py b/src/plassembler/__init__.py index 5bc0476..46708f5 100644 --- a/src/plassembler/__init__.py +++ b/src/plassembler/__init__.py @@ -491,7 +491,7 @@ def run( longreads, Path(f"{outdir}/chopper_long_reads.fastq"), ) - gzip_file(Path(f"{outdir}/chopper_long_reads.fastq")) + gzip_file(Path(f"{outdir}/chopper_long_reads.fastq"), threads) remove_file(Path(f"{outdir}/chopper_long_reads.fastq")) # Raven for long only or '--use_raven' @@ -1103,7 +1103,7 @@ def assembled( longreads, Path(f"{outdir}/chopper_long_reads.fastq"), ) - gzip_file(Path(f"{outdir}/chopper_long_reads.fastq")) + gzip_file(Path(f"{outdir}/chopper_long_reads.fastq"), threads) remove_file(Path(f"{outdir}/chopper_long_reads.fastq")) if short_flag is True: @@ -1454,7 +1454,7 @@ def long( longreads, Path(f"{outdir}/chopper_long_reads.fastq"), ) - gzip_file(Path(f"{outdir}/chopper_long_reads.fastq")) + gzip_file(Path(f"{outdir}/chopper_long_reads.fastq"), threads) remove_file(Path(f"{outdir}/chopper_long_reads.fastq")) # flye - skip directory an option here diff --git a/src/plassembler/utils/concat.py b/src/plassembler/utils/concat.py index 07c3e8d..dba756c 100644 --- a/src/plassembler/utils/concat.py +++ b/src/plassembler/utils/concat.py @@ -1,9 +1,12 @@ import gzip +import os from pathlib import Path -from Bio import SeqIO from loguru import logger +# copied between files in 1 MiB blocks +COPY_BLOCK = 1024 * 1024 + def concatenate_short_fastqs(out_dir): """moves and copies files @@ -32,49 +35,94 @@ def concatenate_short_fastqs(out_dir): logger.error("Error with concatenate_fastqs\n") +def _append_file(source: Path, out_handle, record_marker: bytes): + """Append source to an open binary handle, decompressing it if gzipped. + + Copies in fixed blocks, so memory does not depend on file size, and ensures + the appended block ends with a newline: without that, a source whose final + line is unterminated would run into the next file's first header. + + A non-empty file must start with record_marker (b"@" for fastq, b">" for + fasta). Parsing every record through BioPython used to catch a wrong-format + input; this keeps that check at O(1) instead of O(file). + + :param source: file to append + :param out_handle: destination, opened in binary mode + :param record_marker: first byte every record of this format starts with + :raises ValueError: if source is non-empty and does not start with the marker + """ + opener = gzip.open if Path(source).suffix == ".gz" else open + last_byte = b"" + first = True + with opener(source, "rb") as in_handle: + while True: + block = in_handle.read(COPY_BLOCK) + if not block: + break + if first: + if block[:1] != record_marker: + raise ValueError( + f"{source} does not look like a " + f"{record_marker.decode()}-delimited file: it starts with " + f"{block[:1]!r}" + ) + first = False + out_handle.write(block) + last_byte = block[-1:] + # an empty file contributes nothing, which is normal (e.g. no unmapped reads) + if last_byte and last_byte != b"\n": + out_handle.write(b"\n") + + +def _concatenate(sources, out_path, record_marker: bytes): + """Block-copy sources into out_path, leaving no output behind on failure. + + The copy goes to a sibling .tmp and is renamed into place only once every + source has been read, so a wrong-format second input, a truncated gzip or a + full disk cannot leave a half-written FASTQ/FASTA where the run expects a + complete one. The SeqIO version got this for free by parsing everything + before it opened the output. + + :param sources: files to copy, in order + :param out_path: destination path + :param record_marker: first byte every record of this format starts with + """ + out_path = Path(out_path) + tmp_path = out_path.with_name(f"{out_path.name}.tmp") + try: + with open(tmp_path, "wb") as out_handle: + for source in sources: + _append_file(source, out_handle, record_marker) + except BaseException: + tmp_path.unlink(missing_ok=True) + raise + # same directory, so this is an atomic replace rather than a copy + os.replace(tmp_path, out_path) + + def concatenate_single_fastq(fastq_in1: Path, fastq_in2: Path, fastq_out: Path): """concatenates 2 fastq files + + Concatenating reads needs no parsing. Round-tripping them through + SeqIO.parse into a list of SeqRecords cost roughly 5-10x the file size in + RAM. Only the non-chromosomal short reads pass through here, so that is a + transient spike of a GB or two on a typical isolate rather than the whole + run - but it landed immediately before Unicycler, which needs the memory + itself. A block copy is constant-memory and far faster. + :param fastq_in1: fastq_in1 input fastq 1 :param fastq_in2: fastq_in1 input fastq 2 :param fastq_out: fastq_out output fastq 2 - :param logger: logger :return: """ - - records = [] - - # Read and append records from the first FASTQ file - if fastq_in1.suffix == ".gz": - with gzip.open(fastq_in1, "rt") as handle: - records.extend(SeqIO.parse(handle, "fastq")) - else: - with open(fastq_in1, "r") as handle: - records.extend(SeqIO.parse(handle, "fastq")) - - # Read and append records from the second FASTQ file - if fastq_in2.suffix == ".gz": - with gzip.open(fastq_in2, "rt") as handle: - records.extend(SeqIO.parse(handle, "fastq")) - else: - with open(fastq_in2, "r") as handle: - records.extend(SeqIO.parse(handle, "fastq")) - - # Write the concatenated records to the output FASTQ file - with open(fastq_out, "w") as handle: - SeqIO.write(records, handle, "fastq") + _concatenate([fastq_in1, fastq_in2], fastq_out, b"@") def concatenate_single_fasta(file1: Path, file2: Path, output_file: Path): - sequences = [] - - # Read sequences from the first file - with open(file1, "r") as f1: - sequences.extend(SeqIO.parse(f1, "fasta")) - - # Read sequences from the second file - with open(file2, "r") as f2: - sequences.extend(SeqIO.parse(f2, "fasta")) - - # Write concatenated sequences to the output file - with open(output_file, "w") as output: - SeqIO.write(sequences, output, "fasta") + """concatenates 2 fasta files + :param file1: input fasta 1 + :param file2: input fasta 2 + :param output_file: output fasta + :return: + """ + _concatenate([file1, file2], output_file, b">") diff --git a/src/plassembler/utils/input_commands.py b/src/plassembler/utils/input_commands.py index 25bf9c4..16022ca 100644 --- a/src/plassembler/utils/input_commands.py +++ b/src/plassembler/utils/input_commands.py @@ -188,6 +188,23 @@ def validate_flye_assembly_info(flye_assembly, flye_info): return skip_assembly +# plassembler passes `--trim-approach fixed-crop`, which chopper only gained in +# v0.11.0; older versions reject the flag outright +MIN_CHOPPER_VERSION = (0, 11, 0) + + +def parse_chopper_version(version_output: str): + """Extract (major, minor, patch) from `chopper --version` output. + + :param version_output: stdout of ``chopper --version``, e.g. "chopper 0.11.0". + :return: the version as a tuple of ints, or None if it cannot be parsed. + """ + match = re.search(r"(\d+)\.(\d+)\.(\d+)", version_output) + if match is None: + return None + return tuple(int(part) for part in match.groups()) + + def parse_unicycler_version(version_output: str): """Extract (major, minor, patch) from `unicycler --version` output. @@ -330,6 +347,8 @@ def check_dependencies(): logger.error("fastp not found.") # chopper + # bound up front: the gate below still runs if the version could not be read + chopper_version = "" try: process = sp.Popen(["chopper", "--version"], stdout=sp.PIPE, stderr=sp.PIPE) chopper_out, _ = process.communicate() @@ -340,6 +359,21 @@ def check_dependencies(): except Exception: logger.error("chopper not found.") + # a version pin only binds at install time, so an environment built before the + # bump can still hold a chopper that rejects --trim-approach. Warn rather than + # exit, so anyone deliberately on an older chopper can still run + parsed_chopper_version = parse_chopper_version(chopper_version) + min_chopper_version = ".".join(str(part) for part in MIN_CHOPPER_VERSION) + if parsed_chopper_version is None: + message = f"Could not determine the chopper version from '{chopper_version}'. Plassembler needs chopper >=v{min_chopper_version}." + logger.warning(message) + elif parsed_chopper_version < MIN_CHOPPER_VERSION: + message = f"chopper v{chopper_version} is older than v{min_chopper_version} and will reject the --trim-approach flag Plassembler passes, so long read filtering will fail. Please update chopper, see instructions at https://github.com/gbouras13/plassembler." + logger.warning(message) + else: + message = "chopper version is ok." + logger.info(message) + # mash try: process = sp.Popen(["mash", "version"], stdout=sp.PIPE, stderr=sp.PIPE) diff --git a/src/plassembler/utils/plass_class.py b/src/plassembler/utils/plass_class.py index b8e7318..d9d4d32 100644 --- a/src/plassembler/utils/plass_class.py +++ b/src/plassembler/utils/plass_class.py @@ -1,4 +1,5 @@ import os +import sys from pathlib import Path import pandas as pd @@ -7,6 +8,7 @@ from loguru import logger from plassembler.utils.bam import sam_to_sorted_bam +from plassembler.utils.concat import concatenate_single_fasta from plassembler.utils.depth import ( collate_depths, combine_depth_dfs, @@ -466,21 +468,9 @@ def get_depth_long(self, logdir, pacbio_model, threads, plas_fasta): sam_file: Path = Path(outdir) / "combined_long.sam" sorted_bam: Path = Path(outdir) / "combined_sorted_long.bam" - # # write to combined fasta - - # Create a list to hold the combined sequences - combined_sequences = [] - - # Read and append sequences from the first FASTA file - for record in SeqIO.parse(chromosome, "fasta"): - combined_sequences.append(record) - - # Read and append sequences from the second FASTA file - for record in SeqIO.parse(plas_fasta, "fasta"): - combined_sequences.append(record) - - # Write the combined sequences to the output file - SeqIO.write(combined_sequences, combined_fasta, "fasta") + # write to combined fasta. No parsing needed - the contigs are already + # named as they should be, so this is a plain concatenation + concatenate_single_fasta(chromosome, plas_fasta, combined_fasta) # map minimap_long_reads( @@ -929,32 +919,49 @@ def combine_input_fastas(self, chromosome_fasta: Path, plasmids_fasta: Path): """ # combined input fasta combined_fasta = Path(self.outdir) / "combined.fasta" - chromosome_name = "" + # None, not "", so that a record with an empty header still counts as found + chromosome_name = None - # rename the first contig as chromosome + # rename the first contig as chromosome. Records are streamed rather than + # collected into a list first, so an assembly is never held whole in RAM with open(chromosome_fasta, "r") as f_in, open(combined_fasta, "w") as f_out: - # Parse the input FASTA file - records = list(SeqIO.parse(f_in, "fasta")) - # keep chromosome name - chromosome_name = records[0].id - # Rename the first record - records[0].id = "chromosome" - records[0].description = "" - # Write the modified records to the output FASTA file - SeqIO.write(records, f_out, "fasta") + for index, record in enumerate(SeqIO.parse(f_in, "fasta")): + if index == 0: + # keep chromosome name + chromosome_name = record.id + record.id = "chromosome" + record.description = "" + SeqIO.write(record, f_out, "fasta") + + if chromosome_name is None: + # streaming has no equivalent of list(SeqIO.parse(...))[0] raising + # IndexError, so an empty or non-FASTA chromosome would otherwise + # sail through and leave every depth and copy number computed + # against a combined.fasta with no chromosome in it + logger.error( + f"No contigs were found in {chromosome_fasta}. " + "Please check the chromosome assembly." + ) + # logger.error exits under the CLI's ERROR sink, but not when + # plass_class is driven as a library + sys.exit(1) plasmid_names = [] with open(plasmids_fasta, "r") as f_in, open(combined_fasta, "a") as f_out: - records = list(SeqIO.parse(f_in, "fasta")) - i = 0 - for record in records: - i += 1 + for i, record in enumerate(SeqIO.parse(f_in, "fasta"), start=1): # keep plasmid name plasmid_names.append(record.id) record.id = str(i) - records[0].description = "" - # Write the records to the output FASTA file + if i == 1: + # NOTE: preserved as-is. The original cleared + # records[0].description on every iteration, so only the + # *first* plasmid ever lost its description while the rest + # kept theirs. That looks like a copy-paste slip, and it + # matters because get_contig_circularity() looks for + # "circular" in the description - but changing it would + # change results, so it is left for a separate fix. + record.description = "" SeqIO.write(record, f_out, "fasta") self.chromosome_name = chromosome_name diff --git a/src/plassembler/utils/qc.py b/src/plassembler/utils/qc.py index 16a9957..b66b2e3 100644 --- a/src/plassembler/utils/qc.py +++ b/src/plassembler/utils/qc.py @@ -1,6 +1,8 @@ import gzip import shutil import subprocess as sp +import sys +from contextlib import ExitStack from pathlib import Path from loguru import logger @@ -8,6 +10,80 @@ from plassembler.utils.external_tools import ExternalTool +def gzip_compressor_cmd(threads): + """Command that reads plain data on stdin and writes a gzip stream to stdout. + + Prefers bgzip, which ships with htslib/samtools (already a hard plassembler + dependency) and compresses in parallel. BGZF is a valid gzip stream, so every + downstream reader - flye, minimap2, chopper, gunzip, python's gzip module - + handles the result unchanged. + + Serial gzip dominated the chopper stage: on a 300 MB ONT fastq the chain took + 47.7s, of which only 5.4s was gunzip+chopper and 42s was gzip. + + :param threads: thread count (str or int) to hand to bgzip + :return: argv list for the compressor + """ + if shutil.which("bgzip"): + return ["bgzip", "-@", str(threads), "-c"] + # bgzip should always be present, but never fail QC over a missing binary + return ["gzip"] + + +def _chopper_stderr_tail(err_log, max_lines=20): + """Last few lines of a chopper logfile, ready to embed in an error message. + + Whatever actually went wrong - an unrecognised flag, a bad value - is written + by chopper to its logfile and nowhere else, and users rarely open it. Reading + it back puts the real diagnosis where it will be seen. + + :param err_log: path of the chopper .err logfile + :param max_lines: how many trailing lines to keep + :return: the trailing lines, or "" if the log is missing or empty + """ + try: + log_text = Path(err_log).read_text(errors="replace") + except OSError: + return "" + return "\n".join(log_text.strip().splitlines()[-max_lines:]) + + +def _fastq_has_reads(filtered_long_reads): + """Whether a gzipped fastq actually holds any reads. + + A chopper that dies still leaves a well formed - but empty - gzip member + behind, written by the compressor at the end of the chain, so neither the + file existing nor it being valid gzip says anything about the run. + + :param filtered_long_reads: path of the gzipped fastq to inspect + :return: True if at least one byte of reads can be read back + """ + try: + with gzip.open(filtered_long_reads, "rb") as fh: + return bool(fh.read(1)) + except (OSError, EOFError): + return False + + +def _fail_chopper(reason, err_log): + """Reports a fatal chopper problem and stops the run. + + Everything goes into a single `logger.error`: under the CLI that sink exits + the process, so a second call would never be reached. + + :param reason: what went wrong, e.g. "chopper (return code 2)" + :param err_log: path of the chopper .err logfile + """ + message = f"Error with chopper: {reason}. Please check {err_log}" + stderr_tail = _chopper_stderr_tail(err_log) + if stderr_tail: + message = f"{message}\nchopper stderr:\n{stderr_tail}" + logger.error(message) + # the CLI exits inside logger.error above; this covers qc being driven as a + # library, where no ERROR sink is installed + sys.exit(1) + + def chopper( input_long_reads, outdir, min_length, min_quality, gzip_flag, threads, logdir ): @@ -34,34 +110,81 @@ def chopper( threads, "-l", min_length, + # chopper >=0.11.0 only applies --headcrop/--tailcrop under this approach; + # without it they are silently ignored + "--trim-approach", + "fixed-crop", "--headcrop", "75", "--tailcrop", "75", ] - # `with` guarantees the log and output handles are closed even on error; - # locals are named *_proc to avoid shadowing the `gzip` module / `chopper` - # function name - with open(f"{logfile_prefix}.err", "w") as err_log, open( - filtered_long_reads, "wb" - ) as f: + compressor_cmd = gzip_compressor_cmd(threads) + + # ExitStack guarantees the log, output and pipe handles are closed even if a + # Popen raises partway through building the chain + with ExitStack() as stack: + err_log = stack.enter_context(open(f"{logfile_prefix}.err", "w")) + out_fh = stack.enter_context(open(filtered_long_reads, "wb")) + + stages = [] try: if gzip_flag is True: source_proc = sp.Popen( - ["gunzip", "-c", input_long_reads], stdout=sp.PIPE + ["gunzip", "-c", input_long_reads], stdout=sp.PIPE, stderr=err_log ) + stages.append(("gunzip", source_proc)) + chopper_stdin = source_proc.stdout else: - source_proc = sp.Popen(["cat", input_long_reads], stdout=sp.PIPE) + # plain fastq needs no decompressor: hand the file straight to + # chopper rather than spawning a `cat` to copy it through a pipe + chopper_stdin = stack.enter_context(open(input_long_reads, "rb")) + chopper_proc = sp.Popen( - chopper_cmd, - stdin=source_proc.stdout, - stdout=sp.PIPE, - stderr=err_log, + chopper_cmd, stdin=chopper_stdin, stdout=sp.PIPE, stderr=err_log ) - gzip_proc = sp.Popen(["gzip"], stdin=chopper_proc.stdout, stdout=f) - gzip_proc.communicate() - except Exception: - logger.error("Error with chopper") + stages.append(("chopper", chopper_proc)) + # the parent must drop its copy of each upstream read end, otherwise + # the downstream stage never sees EOF + if gzip_flag is True: + source_proc.stdout.close() + + compress_proc = sp.Popen( + compressor_cmd, stdin=chopper_proc.stdout, stdout=out_fh, stderr=err_log + ) + stages.append((compressor_cmd[0], compress_proc)) + chopper_proc.stdout.close() + except OSError as e: + for _, proc in stages: + proc.kill() + # reap it too, or the error path leaks the zombies the rest of + # this function exists to avoid + proc.wait() + # _fail_chopper exits, so the stages killed above are never reached + # by the wait loop below and reported a second time + _fail_chopper(str(e), f"{logfile_prefix}.err") + + # every stage must be waited on. Previously only the last one was, so a + # failing chopper was silently ignored and left a zombie behind + failures = [] + for name, proc in reversed(stages): + if proc.wait() != 0: + failures.append(f"{name} (return code {proc.returncode})") + + if failures: + _fail_chopper(", ".join(reversed(failures)), f"{logfile_prefix}.err") + + # a dead stage is not the only way to end up with nothing: whatever the + # cause, an empty file here would flow on into Flye/Raven as a zero-read + # assembly, so refuse to hand one over + if not _fastq_has_reads(filtered_long_reads): + _fail_chopper( + f"no reads survived filtering. Check that {input_long_reads} holds " + f"reads longer than {min_length}bp once 150bp of cropping is applied, " + f"with quality above Q{min_quality}", + f"{logfile_prefix}.err", + ) + logger.info("Finished running chopper") @@ -104,10 +227,27 @@ def copy_sr_fastq_file(infile: Path, outfile: Path): logger.error("Error with copy_sr_fastq_file") -def gzip_file(input_path): +def gzip_file(input_path, threads=1): + """gzips a file, in parallel where bgzip is available + + Used by --skip_qc to compress the copied long reads. python's gzip module is + both single threaded and slower than the gzip binary, which is a poor fit for + a multi-GB ONT fastq; fall back to it only if spawning the compressor fails. + + :param input_path: file to compress + :param threads: threads to give the compressor + :return: path of the compressed file + """ input_path = Path(input_path) output_path = input_path.with_suffix(input_path.suffix + ".gz") + try: + with open(input_path, "rb") as f_in, open(output_path, "wb") as f_out: + sp.run(gzip_compressor_cmd(threads), stdin=f_in, stdout=f_out, check=True) + return output_path + except (OSError, sp.CalledProcessError) as e: + logger.warning(f"Falling back to python gzip for {input_path}: {e}") + with open(input_path, "rb") as f_in: with gzip.open(output_path, "wb") as f_out: shutil.copyfileobj(f_in, f_out) diff --git a/tests/test_concat.py b/tests/test_concat.py new file mode 100644 index 0000000..05299c8 --- /dev/null +++ b/tests/test_concat.py @@ -0,0 +1,144 @@ +"""Tests for the streaming concatenation helpers in plassembler.utils.concat.""" + +import gzip + +import pytest +from Bio import SeqIO + +from src.plassembler.utils.concat import ( + concatenate_single_fasta, + concatenate_single_fastq, +) + +READS_A = "@r1\nACGT\n+\nIIII\n@r2\nTTTT\n+\nJJJJ\n" +READS_B = "@r3\nGGGG\n+\nKKKK\n" + + +def records(path): + return [ + (r.id, str(r.seq), tuple(r.letter_annotations["phred_quality"])) + for r in SeqIO.parse(path, "fastq") + ] + + +def test_concatenate_fastq_keeps_every_record(tmp_path): + a, b = tmp_path / "a.fastq", tmp_path / "b.fastq" + a.write_text(READS_A) + b.write_text(READS_B) + out = tmp_path / "out.fastq" + concatenate_single_fastq(a, b, out) + assert [r[0] for r in records(out)] == ["r1", "r2", "r3"] + assert out.read_text() == READS_A + READS_B + + +def test_concatenate_fastq_handles_gzipped_input(tmp_path): + """.gz inputs are decompressed, as the SeqIO version did.""" + a, b = tmp_path / "a.fastq.gz", tmp_path / "b.fastq" + with gzip.open(a, "wt") as fh: + fh.write(READS_A) + b.write_text(READS_B) + out = tmp_path / "out.fastq" + concatenate_single_fastq(a, b, out) + assert [r[0] for r in records(out)] == ["r1", "r2", "r3"] + + +def test_concatenate_fastq_with_empty_first_file(tmp_path): + """An empty input is normal (e.g. no unmapped reads) and contributes nothing.""" + a, b = tmp_path / "a.fastq", tmp_path / "b.fastq" + a.write_text("") + b.write_text(READS_B) + out = tmp_path / "out.fastq" + concatenate_single_fastq(a, b, out) + assert out.read_text() == READS_B + + +def test_concatenate_inserts_missing_newline_between_files(tmp_path): + """A source with no trailing newline must not run into the next file's header.""" + a, b = tmp_path / "a.fastq", tmp_path / "b.fastq" + a.write_text(READS_A.rstrip("\n")) + b.write_text(READS_B) + out = tmp_path / "out.fastq" + concatenate_single_fastq(a, b, out) + assert [r[0] for r in records(out)] == ["r1", "r2", "r3"] + + +def test_concatenate_fastq_rejects_a_fasta(tmp_path): + """Wrong-format input is still caught, as BioPython's parser used to.""" + a, b = tmp_path / "a.fasta", tmp_path / "b.fastq" + a.write_text(">contig1\nACGT\n") + b.write_text(READS_B) + with pytest.raises(ValueError): + concatenate_single_fastq(a, b, tmp_path / "out.fastq") + + +def test_concatenate_fasta_keeps_every_contig(tmp_path): + a, b = tmp_path / "a.fasta", tmp_path / "b.fasta" + a.write_text(">chromosome\nACGTACGT\n") + b.write_text(">1 circular=True\nGGGG\n>2\nTTTT\n") + out = tmp_path / "out.fasta" + concatenate_single_fasta(a, b, out) + parsed = list(SeqIO.parse(out, "fasta")) + assert [r.id for r in parsed] == ["chromosome", "1", "2"] + # descriptions must survive: get_contig_circularity looks for "circular" + assert "circular" in parsed[1].description + + +def test_concatenate_fasta_rejects_a_fastq(tmp_path): + a, b = tmp_path / "a.fastq", tmp_path / "b.fasta" + a.write_text(READS_A) + b.write_text(">1\nACGT\n") + with pytest.raises(ValueError): + concatenate_single_fasta(a, b, tmp_path / "out.fasta") + + +def test_concatenate_leaves_no_output_when_second_file_is_wrong_format(tmp_path): + """A failure must not leave a half-written file where the run expects a whole one. + + The block copy opens the output before it has seen the second input, so + without the .tmp-and-rename the first file's reads would be left behind as a + complete-looking FASTQ. + """ + a, b = tmp_path / "a.fastq", tmp_path / "b.fasta" + a.write_text(READS_A) + b.write_text(">contig1\nACGT\n") + out = tmp_path / "out.fastq" + with pytest.raises(ValueError): + concatenate_single_fastq(a, b, out) + assert not out.exists() + assert list(tmp_path.glob("*.tmp")) == [] + + +def test_concatenate_leaves_an_existing_output_untouched_on_failure(tmp_path): + """The previous output survives a failed re-run rather than being truncated.""" + a, b = tmp_path / "a.fastq", tmp_path / "b.fasta" + a.write_text(READS_A) + b.write_text(">contig1\nACGT\n") + out = tmp_path / "out.fastq" + out.write_text(READS_B) + with pytest.raises(ValueError): + concatenate_single_fastq(a, b, out) + assert out.read_text() == READS_B + + +def test_concatenate_leaves_no_output_when_gzip_is_truncated(tmp_path): + """A truncated .gz raises from the decompressor mid-copy, so nothing is kept.""" + a, b = tmp_path / "a.fastq.gz", tmp_path / "b.fastq" + with gzip.open(a, "wb") as fh: + fh.write((READS_A * 5000).encode()) + raw = a.read_bytes() + a.write_bytes(raw[: len(raw) // 2]) + b.write_text(READS_B) + out = tmp_path / "out.fastq" + with pytest.raises(EOFError): + concatenate_single_fastq(a, b, out) + assert not out.exists() + + +def test_concatenate_accepts_a_string_output_path(tmp_path): + """tests/test_plassembler.py passes os.path.join(...) rather than a Path.""" + a, b = tmp_path / "a.fastq", tmp_path / "b.fastq" + a.write_text(READS_A) + b.write_text(READS_B) + out = str(tmp_path / "out.fastq") + concatenate_single_fastq(a, b, out) + assert [r[0] for r in records(out)] == ["r1", "r2", "r3"] diff --git a/tests/test_data/end_to_end/input_half.fastq.gz b/tests/test_data/end_to_end/input_half.fastq.gz index e7ff299..6d1ab55 100644 Binary files a/tests/test_data/end_to_end/input_half.fastq.gz and b/tests/test_data/end_to_end/input_half.fastq.gz differ diff --git a/tests/test_plass_class.py b/tests/test_plass_class.py index 46b3b58..a33f613 100644 --- a/tests/test_plass_class.py +++ b/tests/test_plass_class.py @@ -125,6 +125,51 @@ def test_combine_input_fastas_good(self): assembly.combine_input_fastas(chrom_fasta, plasmid_fasta) self.assertEqual(expected_return, True) + def test_combine_input_fastas_empty_chromosome(self): + """An empty chromosome.fasta must stop the run, not be combined silently. + + list(SeqIO.parse(...))[0] used to raise IndexError here. Streaming the + records has no equivalent, so the check is explicit - otherwise every + depth and copy number would be computed against a combined.fasta with + no chromosome in it. + """ + assembly = Assembly() + with tempfile.TemporaryDirectory() as tmp: + workdir = Path(tmp) + assembly.outdir = workdir + chrom_fasta = workdir / "chromosome.fasta" + chrom_fasta.write_text("") + plasmid_fasta = workdir / "plasmid.fasta" + plasmid_fasta.write_text(">1 circular=true\nACGT\n") + with self.assertRaises(SystemExit): + assembly.combine_input_fastas(chrom_fasta, plasmid_fasta) + + def test_combine_input_fastas_headers(self): + """Pins the renaming, including the first plasmid losing its description. + + Clearing only plasmid 1's description is a long-standing quirk, and + get_contig_circularity() reads circularity out of that description. It + is asserted here so that fixing it has to be a deliberate change. + """ + assembly = Assembly() + with tempfile.TemporaryDirectory() as tmp: + workdir = Path(tmp) + assembly.outdir = workdir + chrom_fasta = workdir / "chromosome.fasta" + chrom_fasta.write_text(">contig_1 len=5000000\nACGTACGTAC\n") + plasmid_fasta = workdir / "plasmid.fasta" + plasmid_fasta.write_text( + ">1 circular=true\nGGGGGGGGGG\n>2 circular=true\nCCCCCCCCCC\n" + ) + assembly.combine_input_fastas(chrom_fasta, plasmid_fasta) + headers = [ + line.strip() + for line in (workdir / "combined.fasta").read_text().splitlines() + if line.startswith(">") + ] + self.assertEqual(headers, [">chromosome", ">1", ">2 circular=true"]) + self.assertEqual(assembly.chromosome_name, "contig_1") + @pytest.mark.slow def test_check_get_depth(self): expected = True diff --git a/tests/test_plassembler.py b/tests/test_plassembler.py index 4d63a23..2bea6d3 100644 --- a/tests/test_plassembler.py +++ b/tests/test_plassembler.py @@ -30,7 +30,9 @@ # import functions from src.plassembler.utils.input_commands import ( + MIN_CHOPPER_VERSION, check_dependencies, + parse_chopper_version, parse_unicycler_version, validate_fasta, validate_fastas_assembled_mode, @@ -199,6 +201,23 @@ def test_parse_unicycler_version_missing_raises(self): with self.assertRaises(ValueError): parse_unicycler_version("bash: unicycler: command not found\n") + # chopper version parsing + def test_parse_chopper_version_clean(self): + self.assertEqual(parse_chopper_version("chopper 0.11.0\n"), (0, 11, 0)) + + def test_parse_chopper_version_unparseable(self): + # check_dependencies warns on None rather than crashing, so a missing or + # reworded version string must come back as None, not raise + self.assertIsNone(parse_chopper_version("chopper not found")) + self.assertIsNone(parse_chopper_version("")) + + def test_chopper_minimum_version_ordering(self): + # plassembler passes --trim-approach, added in chopper v0.11.0 + self.assertLess(parse_chopper_version("chopper 0.10.0"), MIN_CHOPPER_VERSION) + self.assertGreaterEqual( + parse_chopper_version("chopper 0.13.0"), MIN_CHOPPER_VERSION + ) + # checks all external dependencies are installed @pytest.mark.slow def test_deps(self): diff --git a/tests/test_qc.py b/tests/test_qc.py new file mode 100644 index 0000000..e9e1ffa --- /dev/null +++ b/tests/test_qc.py @@ -0,0 +1,243 @@ +"""Tests for the read QC helpers in plassembler.utils.qc.""" + +import gzip +import os +import shutil +import sys +from pathlib import Path + +import pytest +from loguru import logger + +from src.plassembler.utils.qc import chopper, gzip_compressor_cmd, gzip_file + +TEST_DATA = Path("tests/test_data") + + +@pytest.fixture +def captured_errors(): + """Collect ERROR messages, without the exiting sink conftest installs. + + That sink raises SystemExit, which would stop any later sink - including this + one - from ever seeing the message. The chopper error path exits on its own, + so dropping the sinks for the duration still exercises the real control flow. + """ + logger.remove() + messages = [] + logger.add(messages.append, level="ERROR") + try: + yield messages + finally: + logger.remove() + logger.add(sys.stderr) + logger.add(lambda _: sys.exit(1), level="ERROR") + + +def fake_chopper(tmp_path, monkeypatch, script): + """Put a stub `chopper` first on PATH. + + Lets the failure paths be driven without the real binary, and without the + test depending on which arguments a given chopper release happens to reject. + """ + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + stub = bin_dir / "chopper" + stub.write_text(script) + stub.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}") + + +def test_gzip_compressor_cmd_prefers_bgzip(monkeypatch): + """bgzip ships with samtools (a hard dependency) and is multithreaded.""" + monkeypatch.setattr(shutil, "which", lambda name: "/usr/bin/bgzip") + assert gzip_compressor_cmd(8) == ["bgzip", "-@", "8", "-c"] + + +def test_gzip_compressor_cmd_falls_back_to_gzip(monkeypatch): + """A missing bgzip must degrade to serial gzip, not fail QC.""" + monkeypatch.setattr(shutil, "which", lambda name: None) + assert gzip_compressor_cmd(8) == ["gzip"] + + +@pytest.mark.requires_tool +@pytest.mark.skipif(shutil.which("bgzip") is None, reason="bgzip not installed") +def test_gzip_file_roundtrips(tmp_path): + """bgzip output is a valid gzip stream that python's gzip module reads back.""" + plain = tmp_path / "reads.fastq" + payload = "".join(f"@r{i}\nACGT\n+\nIIII\n" for i in range(1000)) + plain.write_text(payload) + + out = gzip_file(plain, threads=2) + + assert out == tmp_path / "reads.fastq.gz" + with gzip.open(out, "rt") as fh: + assert fh.read() == payload + + +@pytest.mark.requires_tool +@pytest.mark.skipif( + shutil.which("chopper") is None or shutil.which("bgzip") is None, + reason="chopper/bgzip not installed", +) +def test_chopper_output_is_readable_gzip(tmp_path): + """The chopper chain still produces a readable fastq.gz of filtered reads.""" + logdir = tmp_path / "logs" + chopper( + str(TEST_DATA / "test_long.fastq.gz"), + str(tmp_path), + "500", + "9", + True, # gzip_flag + "2", + logdir, + ) + out = tmp_path / "chopper_long_reads.fastq.gz" + assert out.exists() and out.stat().st_size > 0 + with gzip.open(out, "rt") as fh: + lines = fh.readlines() + assert lines, "chopper produced no reads" + assert len(lines) % 4 == 0, "output is not whole fastq records" + assert lines[0].startswith("@") + + +@pytest.mark.requires_tool +@pytest.mark.skipif(shutil.which("chopper") is None, reason="chopper not installed") +def test_chopper_accepts_uncompressed_input(tmp_path): + """Plain fastq input skips the decompressor and is fed to chopper directly.""" + logdir = tmp_path / "logs" + chopper( + str(TEST_DATA / "test_long.fastq"), + str(tmp_path), + "500", + "9", + False, # gzip_flag + "2", + logdir, + ) + out = tmp_path / "chopper_long_reads.fastq.gz" + assert out.exists() and out.stat().st_size > 0 + with gzip.open(out, "rt") as fh: + assert fh.readline().startswith("@") + + +@pytest.mark.requires_tool +@pytest.mark.skipif(shutil.which("chopper") is None, reason="chopper not installed") +def test_chopper_reports_a_failing_stage(tmp_path): + """A stage that exits non-zero must be surfaced. Previously only the final + process was waited on, so a failing chopper was silently swallowed (and left + a zombie behind).""" + logdir = tmp_path / "logs" + with pytest.raises(SystemExit): + # not a gzip file, so the gunzip stage fails + chopper( + str(TEST_DATA / "test_long.fastq"), + str(tmp_path), + "500", + "9", + True, # gzip_flag - wrong for this input, on purpose + "2", + logdir, + ) + + +def test_chopper_failure_is_fatal(tmp_path, monkeypatch, captured_errors): + """A chopper that exits non-zero must stop the run and say why. + + The regression: only the last process in the chain was waited on, so a dead + chopper went unnoticed. The compressor still wrote a valid - but empty - gzip + member, "Finished running chopper" was still logged, and the assemblers were + handed zero reads. The real message only ever reached the logfile. + """ + fake_chopper( + tmp_path, + monkeypatch, + "#!/bin/sh\n" + "echo \"error: unexpected argument '--trim-approach' found\" >&2\n" + "exit 2\n", + ) + logdir = tmp_path / "logs" + + with pytest.raises(SystemExit): + chopper( + str(TEST_DATA / "test_long.fastq"), + str(tmp_path), + "500", + "9", + False, # gzip_flag + "2", + logdir, + ) + + message = "\n".join(captured_errors) + assert "chopper (return code 2)" in message + # chopper's own diagnosis has to reach the user, not just the path to it + assert "unexpected argument '--trim-approach' found" in message + assert str(logdir / "chopper.err") in message + + +def test_chopper_rejects_empty_output(tmp_path, monkeypatch, captured_errors): + """Zero reads is fatal even when every stage exits cleanly. + + An empty fastq.gz is well formed, so nothing downstream notices until Flye or + Raven fails on an assembly with no input. + """ + fake_chopper(tmp_path, monkeypatch, "#!/bin/sh\ncat > /dev/null\nexit 0\n") + logdir = tmp_path / "logs" + + with pytest.raises(SystemExit): + chopper( + str(TEST_DATA / "test_long.fastq"), + str(tmp_path), + "500", + "9", + False, # gzip_flag + "2", + logdir, + ) + + out = tmp_path / "chopper_long_reads.fastq.gz" + assert out.exists(), "the empty file is still written; it must just not be used" + assert "no reads survived filtering" in "\n".join(captured_errors) + + +def test_chopper_missing_binary_is_fatal(tmp_path, monkeypatch, captured_errors): + """A chopper that is not installed at all must exit, not return quietly.""" + monkeypatch.setenv("PATH", str(tmp_path / "empty")) + logdir = tmp_path / "logs" + + with pytest.raises(SystemExit): + chopper( + str(TEST_DATA / "test_long.fastq"), + str(tmp_path), + "500", + "9", + False, # gzip_flag + "2", + logdir, + ) + + assert "Error with chopper" in "\n".join(captured_errors) + + +@pytest.mark.requires_tool +@pytest.mark.skipif(shutil.which("chopper") is None, reason="chopper not installed") +def test_chopper_surfaces_a_real_argument_error(tmp_path, captured_errors): + """The same path against the real binary: clap rejects the value and exits 2.""" + logdir = tmp_path / "logs" + + with pytest.raises(SystemExit): + chopper( + str(TEST_DATA / "test_long.fastq"), + str(tmp_path), + "500", + "NOT_A_NUMBER", # min_quality clap cannot parse + False, # gzip_flag + "2", + logdir, + ) + + message = "\n".join(captured_errors) + assert "chopper (return code 2)" in message + assert "NOT_A_NUMBER" in message, ( + "chopper's stderr must be surfaced, not just its path" + )