From e3ab8bcabb30d0f4a2c98436810243acefd86a99 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Wed, 8 Jul 2026 13:05:43 +0200 Subject: [PATCH 01/13] Exploration of adding deacon --- preprocessing/nextclade/environment.yml | 1 + .../src/loculus_preprocessing/backend.py | 13 +++- .../src/loculus_preprocessing/datatypes.py | 30 ++++++-- .../file_processing_functions.py | 70 +++++++++++++++++-- .../src/loculus_preprocessing/nextclade.py | 8 +-- .../src/loculus_preprocessing/prepro.py | 8 +-- .../nextclade/tests/factory_methods.py | 10 +-- .../tests/test_nextclade_preprocessing.py | 38 ++++++---- 8 files changed, 138 insertions(+), 40 deletions(-) diff --git a/preprocessing/nextclade/environment.yml b/preprocessing/nextclade/environment.yml index 275cbca792..ef8ccccd54 100644 --- a/preprocessing/nextclade/environment.yml +++ b/preprocessing/nextclade/environment.yml @@ -24,3 +24,4 @@ dependencies: - pytest=9.1.1 - pydantic=2.13.4 - diamond=2.2.2 + - deacon=0.15.0 diff --git a/preprocessing/nextclade/src/loculus_preprocessing/backend.py b/preprocessing/nextclade/src/loculus_preprocessing/backend.py index 3c760e5f39..e50fef150f 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/backend.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/backend.py @@ -17,7 +17,7 @@ from .config import Config from .datatypes import ( - FileIdAndName, + FileIdAndNameAndReadUrl, FileUploadInfo, ProcessedEntry, UnprocessedData, @@ -97,7 +97,10 @@ def parse_ndjson(ndjson_data: str) -> Sequence[UnprocessedEntry]: submitted_files = json_object["data"].get("files") file_mapping = ( { - category: [FileIdAndName(fileId=f["fileId"], name=f["name"]) for f in files] + category: [ + FileIdAndNameAndReadUrl(fileId=f["fileId"], name=f["name"], url=f.get("url")) + for f in files + ] for category, files in submitted_files.items() } if submitted_files @@ -245,6 +248,12 @@ def upload_embl_file_to_presigned_url( raise RuntimeError(msg) +def download_file(config: Config, url: str, save_path: str): + response = requests.get(url, timeout=config.backend_request_timeout_seconds) + response.raise_for_status() + Path(save_path).write_bytes(response.content) + + def download_minimizer(config, save_path): if config.minimizer_url: url = config.minimizer_url diff --git a/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py b/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py index f957168f0f..65d88725ee 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py @@ -1,7 +1,8 @@ from collections.abc import Iterable -from dataclasses import dataclass, field +from dataclasses import dataclass, field, fields from enum import StrEnum, unique -from typing import Any, Final +import json +from typing import Any, Final, Self AccessionVersion = str GeneName = str @@ -82,9 +83,10 @@ def from_single(cls, name: str, type, message: str): @dataclass -class FileIdAndName: +class FileIdAndNameAndReadUrl: fileId: str # noqa: N815 name: str + url: str | None = None @dataclass @@ -95,7 +97,7 @@ class UnprocessedData: submissionId: str # noqa: N815 metadata: InputMetadata unalignedNucleotideSequences: dict[SequenceName, NucleotideSequence | None] # noqa: N815 - files: dict[FileCategory, list[FileIdAndName]] | None + files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None @dataclass @@ -111,7 +113,7 @@ class UnprocessedEntry: @dataclass class UnprocessedAfterNextclade: inputMetadata: InputMetadata # noqa: N815 - files: dict[FileCategory, list[FileIdAndName]] | None + files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None # Derived metadata produced by Nextclade nextcladeMetadata: dict[SequenceName, Any] | None # noqa: N815 unalignedNucleotideSequences: dict[SequenceName, NucleotideSequence | None] # noqa: N815 @@ -127,7 +129,7 @@ class UnprocessedAfterNextclade: @dataclass class ProcessedData: metadata: ProcessedMetadata - files: dict[FileCategory, list[FileIdAndName]] | None + files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None unalignedNucleotideSequences: dict[SequenceName, Any] # noqa: N815 alignedNucleotideSequences: dict[SequenceName, Any] # noqa: N815 nucleotideInsertions: dict[SequenceName, Any] # noqa: N815 @@ -265,3 +267,19 @@ class MoleculeType(StrEnum): class Topology(StrEnum): LINEAR = "linear" CIRCULAR = "circular" + + +@dataclass +class DeaconSummary: + time: float + seqs_in: int + seqs_out: int + seqs_removed: int + seqs_removed_proportion: float + + @classmethod + def from_json(cls, json_path: str) -> Self: + with open(json_path, encoding="utf-8") as f: + data = json.load(f) + wanted = {f.name for f in fields(cls)} + return cls(**{k: v for k, v in data.items() if k in wanted}) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index 464659992a..125f3036ff 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -1,17 +1,28 @@ import logging +import os +import subprocess +from tempfile import TemporaryDirectory +from requests import HTTPError + +from loculus_preprocessing.backend import download_file +from loculus_preprocessing.config import Config from loculus_preprocessing.datatypes import ( AnnotationSourceType, + DeaconSummary, FileCategory, - FileIdAndName, + FileIdAndNameAndReadUrl, ProcessingAnnotation, ) +from loculus_preprocessing.processing_functions import _internal_error_message logger = logging.getLogger(__name__) +# TODO: pass Config here as well (backend_request_timeout_seconds, host read proportion) def process_submitted_files( - file_mapping: dict[FileCategory, list[FileIdAndName]], + config: Config, + file_mapping: dict[FileCategory, list[FileIdAndNameAndReadUrl]], ) -> tuple[list[ProcessingAnnotation], list[ProcessingAnnotation]]: errors: list[ProcessingAnnotation] = [] warnings: list[ProcessingAnnotation] = [] @@ -22,7 +33,7 @@ def process_submitted_files( continue match category: case FileCategory.RAW_READS: - rr_errors, rr_warnings = validate_raw_reads_submission(files) + rr_errors, rr_warnings = validate_raw_reads_submission(config, files, 0.05) errors.extend(rr_errors) warnings.extend(rr_warnings) case _: @@ -42,7 +53,9 @@ def process_submitted_files( def validate_raw_reads_submission( - files: list[FileIdAndName], + config: Config, + files: list[FileIdAndNameAndReadUrl], + max_host_proportion: float, ) -> tuple[list[ProcessingAnnotation], list[ProcessingAnnotation]]: errors: list[ProcessingAnnotation] = [] warnings: list[ProcessingAnnotation] = [] @@ -74,4 +87,53 @@ def validate_raw_reads_submission( ) ) + with TemporaryDirectory() as tmp_dir: + for file in files: + if (url := file.url) is None: + message = f"Cannot download file '{file.name}': preprocessing did not receive a URL" + errors.append( + ProcessingAnnotation.from_single( + name=file.name, type=AnnotationSourceType.FILE, message=message + ) + ) + continue + file_name_internal = os.path.join(tmp_dir, file.fileId) + try: + download_file(config, url, file_name_internal) + except HTTPError as e: + logger.error(f"Error downloading file '{file.name}' from S3: {e}") + summary_json = os.path.join(tmp_dir, f"{file.fileId}_summary.json") + deacon_summary = run_deacon(file_name_internal, summary_json) + if deacon_summary.seqs_removed_proportion > (1.0 - max_host_proportion): + message = f"File {file.name} (id: {file.fileId}) had " + errors.append( + ProcessingAnnotation.from_single( + name=file.name, type=AnnotationSourceType.FILE, message=message + ) + ) + return errors, warnings + + +def run_deacon(input_file: str, summary_json: str): + index = "" + args = [ + "deacon", + "filter", + "--threads", + 1, + "--summary", + summary_json, + index, + input_file, + "2> /dev/null", + ] + + logger.debug(f"Checking for host sequences in raw read file {input_file}: {' '.join(args)}") + + exit_code = subprocess.run(args, check=False).returncode # noqa: S603 + if exit_code != 0: + msg = f"deacon filter failed with exit code {exit_code}" + raise Exception(msg) + + return DeaconSummary.from_json(summary_json) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/nextclade.py b/preprocessing/nextclade/src/loculus_preprocessing/nextclade.py index 8545472b63..b15882d13b 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/nextclade.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/nextclade.py @@ -27,7 +27,7 @@ AnnotationSourceType, FastaId, FileCategory, - FileIdAndName, + FileIdAndNameAndReadUrl, GeneName, GenericSequence, NucleotideInsertion, @@ -803,9 +803,9 @@ def enrich_with_nextclade( # noqa: PLR0914 } for entry in unprocessed } - input_files: dict[AccessionVersion, dict[FileCategory, list[FileIdAndName]] | None] = { - entry.accessionVersion: entry.data.files for entry in unprocessed - } + input_files: dict[ + AccessionVersion, dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None + ] = {entry.accessionVersion: entry.data.files for entry in unprocessed} batch = assign_segment_for_alignment(unprocessed, config=config, dataset_dir=dataset_dir) unaligned_nucleotide_sequences = batch.unalignedNucleotideSequences diff --git a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py index 6ea6a7eeb9..884983946e 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py @@ -31,7 +31,7 @@ AnnotationSource, AnnotationSourceType, FileCategory, - FileIdAndName, + FileIdAndNameAndReadUrl, GeneName, InputData, InputMetadata, @@ -537,7 +537,7 @@ def process_single( accession_version, unprocessed, config ) - file_errors, file_warnings = process_submitted_files(unprocessed.files or {}) + file_errors, file_warnings = process_submitted_files(config, unprocessed.files or {}) processed_entry = ProcessedEntry( accession=accession_from_str(accession_version), @@ -592,7 +592,7 @@ def process_single_unaligned( accession_version, unprocessed, config ) - file_errors, file_warnings = process_submitted_files(unprocessed.files or {}) + file_errors, file_warnings = process_submitted_files(config, unprocessed.files or {}) return processed_entry_no_alignment( accession_version=accession_version, @@ -680,7 +680,7 @@ def upload_flatfiles(processed: Sequence[SubmissionData], config: Config) -> Non upload_embl_file_to_presigned_url(file_content, upload_info.url, upload_info.headers) processed_files = submission_data.processed_entry.data.files or {} processed_files.setdefault(FileCategory.ANNOTATIONS, []).append( - FileIdAndName(fileId=file_id, name=file_name) + FileIdAndNameAndReadUrl(fileId=file_id, name=file_name) ) submission_data.processed_entry.data.files = processed_files except Exception as e: diff --git a/preprocessing/nextclade/tests/factory_methods.py b/preprocessing/nextclade/tests/factory_methods.py index 679601e744..02e5f2159c 100644 --- a/preprocessing/nextclade/tests/factory_methods.py +++ b/preprocessing/nextclade/tests/factory_methods.py @@ -10,7 +10,7 @@ AnnotationSource, AnnotationSourceType, FileCategory, - FileIdAndName, + FileIdAndNameAndReadUrl, NucleotideSequence, ProcessedData, ProcessedEntry, @@ -79,7 +79,7 @@ def create_unprocessed_entry( accession_id: str, sequences: dict[SegmentName, NucleotideSequence | None], group_id: int = 2, - files: dict[FileCategory, list[FileIdAndName]] | None = None, + files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None = None, ) -> UnprocessedEntry: return UnprocessedEntry( accessionVersion=f"LOC_{accession_id}.1", @@ -134,7 +134,7 @@ def create_processed_entry( errors: list[ProcessingAnnotation] | None = None, warnings: list[ProcessingAnnotation] | None = None, processed_alignment: ProcessedAlignment | None = None, - files: dict[FileCategory, list[FileIdAndName]] | None = None, + files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None = None, ) -> ProcessedEntry: if errors is None: errors = [] @@ -169,11 +169,11 @@ def create_processed_entry( class Case: name: str input_metadata: dict[str, str | None] = field(default_factory=dict) - input_files: dict[FileCategory, list[FileIdAndName]] | None = None + input_files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None = None input_sequence: dict[str, str | None] = field(default_factory=lambda: {"main": None}) accession_id: str = "000999" expected_metadata: dict[str, ProcessedMetadataValue] = field(default_factory=dict) - expected_files: dict[FileCategory, list[FileIdAndName]] | None = None + expected_files: dict[FileCategory, list[FileIdAndNameAndReadUrl]] | None = None expected_errors: list[ProcessingAnnotation] | None = None expected_warnings: list[ProcessingAnnotation] | None = None expected_processed_alignment: ProcessedAlignment | None = None diff --git a/preprocessing/nextclade/tests/test_nextclade_preprocessing.py b/preprocessing/nextclade/tests/test_nextclade_preprocessing.py index 71352cd060..65ba48726f 100644 --- a/preprocessing/nextclade/tests/test_nextclade_preprocessing.py +++ b/preprocessing/nextclade/tests/test_nextclade_preprocessing.py @@ -27,7 +27,7 @@ from loculus_preprocessing.datatypes import ( AnnotationSourceType, FileCategory, - FileIdAndName, + FileIdAndNameAndReadUrl, SegmentClassificationMethod, SubmissionData, UnprocessedData, @@ -291,8 +291,8 @@ def invalid_sequence() -> str: input_metadata={}, input_files={ FileCategory.RAW_READS: [ - FileIdAndName(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndName(fileId="file-id-0002", name="reads_R2.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), ] }, input_sequence={"fastaHeader": sequence_with_mutation("single")}, @@ -308,8 +308,8 @@ def invalid_sequence() -> str: }, expected_files={ FileCategory.RAW_READS: [ - FileIdAndName(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndName(fileId="file-id-0002", name="reads_R2.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), ] }, expected_errors=[], @@ -331,9 +331,9 @@ def invalid_sequence() -> str: input_metadata={}, input_files={ FileCategory.RAW_READS: [ - FileIdAndName(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndName(fileId="file-id-0002", name="reads_R2.fastq"), - FileIdAndName(fileId="file-id-0003", name="reads_R3.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0003", name="reads_R3.fastq"), ] }, input_sequence={"fastaHeader": sequence_with_mutation("single")}, @@ -349,9 +349,9 @@ def invalid_sequence() -> str: }, expected_files={ FileCategory.RAW_READS: [ - FileIdAndName(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndName(fileId="file-id-0002", name="reads_R2.fastq"), - FileIdAndName(fileId="file-id-0003", name="reads_R3.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), + FileIdAndNameAndReadUrl(fileId="file-id-0003", name="reads_R3.fastq"), ] }, expected_errors=build_processing_annotations( @@ -381,7 +381,9 @@ def invalid_sequence() -> str: name="with unrecognized raw read file extension", input_metadata={}, input_files={ - FileCategory.RAW_READS: [FileIdAndName(fileId="file-id-0001", name="reads.txt")] + FileCategory.RAW_READS: [ + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") + ] }, input_sequence={"fastaHeader": sequence_with_mutation("single")}, accession_id="1", @@ -395,7 +397,9 @@ def invalid_sequence() -> str: "variant": True, }, expected_files={ - FileCategory.RAW_READS: [FileIdAndName(fileId="file-id-0001", name="reads.txt")] + FileCategory.RAW_READS: [ + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") + ] }, expected_errors=build_processing_annotations( [ @@ -1280,10 +1284,14 @@ def invalid_sequence() -> str: }, accession_id="1", input_files={ - FileCategory.RAW_READS: [FileIdAndName(fileId="file-id-0001", name="reads.txt")] + FileCategory.RAW_READS: [ + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") + ] }, expected_files={ - FileCategory.RAW_READS: [FileIdAndName(fileId="file-id-0001", name="reads.txt")] + FileCategory.RAW_READS: [ + FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") + ] }, expected_metadata={ "length_ebola-sudan": len(consensus_sequence("ebola-sudan")), From 3e6b2ba237290b52f4831b656f954247e82acc81 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Thu, 9 Jul 2026 16:10:00 +0200 Subject: [PATCH 02/13] Rename --- .../src/loculus_preprocessing/file_processing_functions.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index 125f3036ff..cd194043fe 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -102,8 +102,8 @@ def validate_raw_reads_submission( download_file(config, url, file_name_internal) except HTTPError as e: logger.error(f"Error downloading file '{file.name}' from S3: {e}") - summary_json = os.path.join(tmp_dir, f"{file.fileId}_summary.json") - deacon_summary = run_deacon(file_name_internal, summary_json) + summary_json_path = os.path.join(tmp_dir, f"{file.fileId}_summary.json") + deacon_summary = run_deacon(file_name_internal, summary_json_path) if deacon_summary.seqs_removed_proportion > (1.0 - max_host_proportion): message = f"File {file.name} (id: {file.fileId}) had " errors.append( From f05471a5452c4c69b045beef630c20d123a09457 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Thu, 9 Jul 2026 16:49:04 +0200 Subject: [PATCH 03/13] More exploring (still rough!) --- .../src/loculus_preprocessing/backend.py | 2 +- .../file_processing_functions.py | 46 +++++++++++-------- 2 files changed, 28 insertions(+), 20 deletions(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/backend.py b/preprocessing/nextclade/src/loculus_preprocessing/backend.py index e50fef150f..7edfcd4f91 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/backend.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/backend.py @@ -248,7 +248,7 @@ def upload_embl_file_to_presigned_url( raise RuntimeError(msg) -def download_file(config: Config, url: str, save_path: str): +def download_file(config: Config, url: str, save_path: str) -> None: response = requests.get(url, timeout=config.backend_request_timeout_seconds) response.raise_for_status() Path(save_path).write_bytes(response.content) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index cd194043fe..c329d1148c 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -14,12 +14,10 @@ FileIdAndNameAndReadUrl, ProcessingAnnotation, ) -from loculus_preprocessing.processing_functions import _internal_error_message logger = logging.getLogger(__name__) -# TODO: pass Config here as well (backend_request_timeout_seconds, host read proportion) def process_submitted_files( config: Config, file_mapping: dict[FileCategory, list[FileIdAndNameAndReadUrl]], @@ -88,6 +86,10 @@ def validate_raw_reads_submission( ) with TemporaryDirectory() as tmp_dir: + index_path = os.path.join(tmp_dir, "panhuman-1.k31w15.idx") + if not run_deacon_fetch(index_path): + return errors, warnings + for file in files: if (url := file.url) is None: message = f"Cannot download file '{file.name}': preprocessing did not receive a URL" @@ -102,10 +104,15 @@ def validate_raw_reads_submission( download_file(config, url, file_name_internal) except HTTPError as e: logger.error(f"Error downloading file '{file.name}' from S3: {e}") + continue summary_json_path = os.path.join(tmp_dir, f"{file.fileId}_summary.json") - deacon_summary = run_deacon(file_name_internal, summary_json_path) - if deacon_summary.seqs_removed_proportion > (1.0 - max_host_proportion): - message = f"File {file.name} (id: {file.fileId}) had " + deacon_summary = run_deacon_filter(index_path, file_name_internal, summary_json_path) + if deacon_summary.seqs_removed_proportion > max_host_proportion: + message = ( + f"File {file.name} (id: {file.fileId}) had a host reads proportion of " + f"{deacon_summary.seqs_removed_proportion}, maximum allowed proportion " + f"is {max_host_proportion}" + ) errors.append( ProcessingAnnotation.from_single( name=file.name, type=AnnotationSourceType.FILE, message=message @@ -115,23 +122,24 @@ def validate_raw_reads_submission( return errors, warnings -def run_deacon(input_file: str, summary_json: str): - index = "" - args = [ - "deacon", - "filter", - "--threads", - 1, - "--summary", - summary_json, - index, - input_file, - "2> /dev/null", - ] +def run_deacon_fetch(index_path: str) -> bool: + args = ["deacon", "index", "fetch", "--output", index_path, "panhuman-1"] + exit_code = subprocess.run(args, check=False, stderr=subprocess.DEVNULL).returncode # noqa: S603 + if exit_code != 0: + msg = f"deacon fetch failed with exit code {exit_code}" + logger.error(msg) + return False + return True + + +def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconSummary: + args = ["deacon", "filter", "--threads", "1", "--summary", summary_json, index, input_file] logger.debug(f"Checking for host sequences in raw read file {input_file}: {' '.join(args)}") - exit_code = subprocess.run(args, check=False).returncode # noqa: S603 + exit_code = subprocess.run( # noqa: S603 + args, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ).returncode if exit_code != 0: msg = f"deacon filter failed with exit code {exit_code}" raise Exception(msg) From 12e5d7e13ef6ebf739e50a20358c58340abaaf85 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Fri, 10 Jul 2026 10:17:19 +0200 Subject: [PATCH 04/13] Pass enabled file categories to prepro config. Download deacon index once at startup --- .../loculus-preprocessing-config.yaml | 8 ++++ .../src/loculus_preprocessing/backend.py | 1 + .../src/loculus_preprocessing/config.py | 2 + .../file_processing_functions.py | 46 +++++++++++-------- .../src/loculus_preprocessing/prepro.py | 28 +++++++++-- 5 files changed, 62 insertions(+), 23 deletions(-) diff --git a/kubernetes/loculus/templates/loculus-preprocessing-config.yaml b/kubernetes/loculus/templates/loculus-preprocessing-config.yaml index dabe36aa3b..7db9039b83 100644 --- a/kubernetes/loculus/templates/loculus-preprocessing-config.yaml +++ b/kubernetes/loculus/templates/loculus-preprocessing-config.yaml @@ -18,6 +18,14 @@ data: {{- if and (hasKey $organismConfig.schema "submissionDataTypes") (hasKey $organismConfig.schema.submissionDataTypes "maxSequencesPerEntry") }} max_sequences_per_entry: {{ $organismConfig.schema.submissionDataTypes.maxSequencesPerEntry }} {{- end }} + {{- if and (hasKey $organismConfig.schema "submissionDataTypes") (hasKey $organismConfig.schema.submissionDataTypes "files") }} + {{- if $organismConfig.schema.submissionDataTypes.files.enabled }} + submission_file_categories: + {{- range $organismConfig.schema.submissionDataTypes.files.categories }} + - {{ .name }} + {{- end }} + {{- end }} + {{- end }} processing_spec: {{- $args := dict "metadata" $metadata "referenceGenomes" $organismConfig.referenceGenomes }} {{- include "loculus.preprocessingSpecs" $args | nindent 6 }} diff --git a/preprocessing/nextclade/src/loculus_preprocessing/backend.py b/preprocessing/nextclade/src/loculus_preprocessing/backend.py index 7edfcd4f91..f7eea9cedc 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/backend.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/backend.py @@ -248,6 +248,7 @@ def upload_embl_file_to_presigned_url( raise RuntimeError(msg) +# TODO: Stream data instead of loading all at once with write_bytes def download_file(config: Config, url: str, save_path: str) -> None: response = requests.get(url, timeout=config.backend_request_timeout_seconds) response.raise_for_status() diff --git a/preprocessing/nextclade/src/loculus_preprocessing/config.py b/preprocessing/nextclade/src/loculus_preprocessing/config.py index cb3757bf27..e4bdf4b4f3 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/config.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/config.py @@ -27,6 +27,7 @@ NEXTCLADE_PREFIX = "nextclade." ASSIGNED_REFERENCE_PREFIX = "ASSIGNED_REFERENCE" INTERNAL_INPUT_PREFIXES = (NEXTCLADE_PREFIX, ASSIGNED_REFERENCE_PREFIX) +DEACON_INDEX = "panhuman-1.k31w15.idx" class EmblInfoMetadataPropertyNames(BaseModel): @@ -120,6 +121,7 @@ class Config(BaseModel): minimizer_url: str | None = None diamond_dmnd_url: str | None = None + submission_file_categories: list[str] = Field(default_factory=list) create_embl_file: bool = False scientific_name: str = "Orthonairovirus haemorrhagiae" molecule_type: MoleculeType = MoleculeType.GENOMIC_RNA diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index c329d1148c..64119b2dfc 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -6,7 +6,7 @@ from requests import HTTPError from loculus_preprocessing.backend import download_file -from loculus_preprocessing.config import Config +from loculus_preprocessing.config import DEACON_INDEX, Config from loculus_preprocessing.datatypes import ( AnnotationSourceType, DeaconSummary, @@ -20,6 +20,7 @@ def process_submitted_files( config: Config, + dataset_dir: str, file_mapping: dict[FileCategory, list[FileIdAndNameAndReadUrl]], ) -> tuple[list[ProcessingAnnotation], list[ProcessingAnnotation]]: errors: list[ProcessingAnnotation] = [] @@ -31,7 +32,9 @@ def process_submitted_files( continue match category: case FileCategory.RAW_READS: - rr_errors, rr_warnings = validate_raw_reads_submission(config, files, 0.05) + rr_errors, rr_warnings = validate_raw_reads_submission( + config, dataset_dir, files, 0.05 + ) errors.extend(rr_errors) warnings.extend(rr_warnings) case _: @@ -52,6 +55,7 @@ def process_submitted_files( def validate_raw_reads_submission( config: Config, + dataset_dir: str, files: list[FileIdAndNameAndReadUrl], max_host_proportion: float, ) -> tuple[list[ProcessingAnnotation], list[ProcessingAnnotation]]: @@ -85,28 +89,33 @@ def validate_raw_reads_submission( ) ) + deacon_index = os.path.join(dataset_dir, DEACON_INDEX) with TemporaryDirectory() as tmp_dir: - index_path = os.path.join(tmp_dir, "panhuman-1.k31w15.idx") - if not run_deacon_fetch(index_path): - return errors, warnings - for file in files: if (url := file.url) is None: - message = f"Cannot download file '{file.name}': preprocessing did not receive a URL" + message = f"No URL for file '{file.name}' with id '{file.fileId}'" errors.append( ProcessingAnnotation.from_single( name=file.name, type=AnnotationSourceType.FILE, message=message ) ) continue - file_name_internal = os.path.join(tmp_dir, file.fileId) + + file_name_internal = os.path.join(tmp_dir, f"{file.fileId}-{file.name}") try: download_file(config, url, file_name_internal) except HTTPError as e: - logger.error(f"Error downloading file '{file.name}' from S3: {e}") + message = f"Error downloading file '{file.name}' from S3: {e}" + logger.error(message) + errors.append( + ProcessingAnnotation.from_single( + name=file.name, type=AnnotationSourceType.FILE, message=message + ) + ) continue + summary_json_path = os.path.join(tmp_dir, f"{file.fileId}_summary.json") - deacon_summary = run_deacon_filter(index_path, file_name_internal, summary_json_path) + deacon_summary = run_deacon_filter(deacon_index, file_name_internal, summary_json_path) if deacon_summary.seqs_removed_proportion > max_host_proportion: message = ( f"File {file.name} (id: {file.fileId}) had a host reads proportion of " @@ -124,24 +133,25 @@ def validate_raw_reads_submission( def run_deacon_fetch(index_path: str) -> bool: args = ["deacon", "index", "fetch", "--output", index_path, "panhuman-1"] - exit_code = subprocess.run(args, check=False, stderr=subprocess.DEVNULL).returncode # noqa: S603 + logger.debug(f"Running Deacon fetch: {args}") + + exit_code = subprocess.run(args, check=False).returncode # noqa: S603 if exit_code != 0: - msg = f"deacon fetch failed with exit code {exit_code}" - logger.error(msg) + message = f"Deacon fetch failed with exit code {exit_code}" + logger.error(message) return False return True def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconSummary: args = ["deacon", "filter", "--threads", "1", "--summary", summary_json, index, input_file] - - logger.debug(f"Checking for host sequences in raw read file {input_file}: {' '.join(args)}") + logger.debug(f"Running Deacon filter on '{input_file}': {args}") exit_code = subprocess.run( # noqa: S603 args, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL ).returncode if exit_code != 0: - msg = f"deacon filter failed with exit code {exit_code}" - raise Exception(msg) - + message = f"Deacon filter failed with exit code {exit_code}" + logger.error(message) + raise Exception(message) return DeaconSummary.from_json(summary_json) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py index 884983946e..cceda1478d 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py @@ -1,11 +1,15 @@ import logging +import os import time from collections import defaultdict from collections.abc import Sequence from tempfile import TemporaryDirectory from typing import Any -from loculus_preprocessing.file_processing_functions import process_submitted_files +from loculus_preprocessing.file_processing_functions import ( + process_submitted_files, + run_deacon_fetch, +) from .backend import ( download_diamond_db, @@ -17,6 +21,7 @@ ) from .config import ( ASSIGNED_REFERENCE_PREFIX, + DEACON_INDEX, METADATA_DEPENDENCY_PREFIX, NEXTCLADE_PREFIX, AlignmentRequirement, @@ -518,6 +523,7 @@ def unpack_annotations(config, nextclade_metadata: dict[str, Any] | None) -> dic def process_single( accession_version: AccessionVersion, unprocessed: UnprocessedAfterNextclade, + dataset_dir: str, config: Config, ) -> SubmissionData: """Process a single sequence per config""" @@ -537,7 +543,9 @@ def process_single( accession_version, unprocessed, config ) - file_errors, file_warnings = process_submitted_files(config, unprocessed.files or {}) + file_errors, file_warnings = process_submitted_files( + config, dataset_dir, unprocessed.files or {} + ) processed_entry = ProcessedEntry( accession=accession_from_str(accession_version), @@ -578,6 +586,7 @@ def process_single( def process_single_unaligned( accession_version: AccessionVersion, unprocessed: UnprocessedData, + dataset_dir: str, config: Config, ) -> SubmissionData: """Process a single sequence per config""" @@ -592,7 +601,9 @@ def process_single_unaligned( accession_version, unprocessed, config ) - file_errors, file_warnings = process_submitted_files(config, unprocessed.files or {}) + file_errors, file_warnings = process_submitted_files( + config, dataset_dir, unprocessed.files or {} + ) return processed_entry_no_alignment( accession_version=accession_version, @@ -646,7 +657,7 @@ def process_all( nextclade_results = enrich_with_nextclade(unprocessed, dataset_dir, config) for id, result in nextclade_results.items(): try: - processed_single = process_single(id, result, config) + processed_single = process_single(id, result, dataset_dir, config) except Exception as e: logger.error(f"Processing failed for {id} with error: {e}") processed_single = processed_entry_with_errors(id) @@ -655,7 +666,7 @@ def process_all( for entry in unprocessed: try: processed_single = process_single_unaligned( - entry.accessionVersion, entry.data, config + entry.accessionVersion, entry.data, dataset_dir, config ) except Exception as e: logger.error(f"Processing failed for {entry.accessionVersion} with error: {e}") @@ -713,6 +724,13 @@ def run(config: Config) -> None: # noqa: C901 msg = "Diamond database URL must be provided for diamond segment classification" raise ValueError(msg) download_diamond_db(config, dataset_dir + "/diamond/diamond.dmnd") + if FileCategory.RAW_READS in config.submission_file_categories: + # TODO: Should we host this ourselves somewhere instead of fetching from external? + index_path = os.path.join(dataset_dir, DEACON_INDEX) + if not run_deacon_fetch(index_path): + msg = "Failed to fetch Deacon index" + raise RuntimeError(msg) + total_processed = 0 etag = None last_force_refresh = time.time() From 6a5b127b50e63da820cd5c0cbbfb893209e5a558 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Fri, 10 Jul 2026 10:33:14 +0200 Subject: [PATCH 05/13] Only run host check if basic checks pass --- .../nextclade/src/loculus_preprocessing/datatypes.py | 2 +- .../src/loculus_preprocessing/file_processing_functions.py | 4 ++++ 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py b/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py index 65d88725ee..8029b67eec 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/datatypes.py @@ -1,7 +1,7 @@ +import json from collections.abc import Iterable from dataclasses import dataclass, field, fields from enum import StrEnum, unique -import json from typing import Any, Final, Self AccessionVersion = str diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index 64119b2dfc..0d7edf06fc 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -89,6 +89,10 @@ def validate_raw_reads_submission( ) ) + if errors: + # only do in-depth validation if basic checks pass + return errors, warnings + deacon_index = os.path.join(dataset_dir, DEACON_INDEX) with TemporaryDirectory() as tmp_dir: for file in files: From 06278f9d3dc9f5d6e79f2bb1e44ac3fc968ca4de Mon Sep 17 00:00:00 2001 From: maverbiest Date: Fri, 10 Jul 2026 11:54:26 +0200 Subject: [PATCH 06/13] Move file processing tests to dedicated file --- .../file_processing_functions.py | 7 +- .../nextclade/tests/test_file_processing.py | 189 +++++++++++++++++ .../tests/test_nextclade_preprocessing.py | 190 ------------------ 3 files changed, 194 insertions(+), 192 deletions(-) create mode 100644 preprocessing/nextclade/tests/test_file_processing.py diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index 0d7edf06fc..f0847de314 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -1,6 +1,6 @@ import logging import os -import subprocess +import subprocess # noqa: S404 from tempfile import TemporaryDirectory from requests import HTTPError @@ -33,7 +33,10 @@ def process_submitted_files( match category: case FileCategory.RAW_READS: rr_errors, rr_warnings = validate_raw_reads_submission( - config, dataset_dir, files, 0.05 + config, + dataset_dir, + files, + 0.05, # TODO: make configurable ) errors.extend(rr_errors) warnings.extend(rr_warnings) diff --git a/preprocessing/nextclade/tests/test_file_processing.py b/preprocessing/nextclade/tests/test_file_processing.py new file mode 100644 index 0000000000..bba2808fef --- /dev/null +++ b/preprocessing/nextclade/tests/test_file_processing.py @@ -0,0 +1,189 @@ +# ruff: noqa: S101 + +from unittest.mock import MagicMock + +import pytest + +from loculus_preprocessing import file_processing_functions, prepro +from loculus_preprocessing.config import Config +from loculus_preprocessing.datatypes import ( + DeaconSummary, + FileCategory, + FileIdAndNameAndReadUrl, +) +from loculus_preprocessing.file_processing_functions import ( + process_submitted_files, + validate_raw_reads_submission, +) + +MAX_HOST_PROPORTION = 0.05 + + +@pytest.fixture +def config() -> Config: + return Config() + + +@pytest.fixture +def mock_deacon(monkeypatch): + """Mock the deacon index file download and deacon filter calls. + + Returns a setter so each test can choose the reported host-read proportion (default 0.0). + """ + monkeypatch.setattr( + file_processing_functions, "download_file", lambda config, url, save_path: None + ) + + def set_removed_proportion(removed_proportion: float = 0.0) -> None: + monkeypatch.setattr( + file_processing_functions, + "run_deacon_filter", + lambda index, input_file, summary_json: deacon_summary(removed_proportion), + ) + + set_removed_proportion(0.0) + return set_removed_proportion + + +def deacon_summary(removed_proportion: float) -> DeaconSummary: + """Build a DeaconSummary reporting the given proportion of removed host reads.""" + seqs_in = 100 + seqs_removed = round(seqs_in * removed_proportion) + return DeaconSummary( + time=0.0, + seqs_in=seqs_in, + seqs_out=seqs_in - seqs_removed, + seqs_removed=seqs_removed, + seqs_removed_proportion=removed_proportion, + ) + + +def fastq_file( + file_id: str, name: str, url: str | None = "https://example.test/reads" +) -> FileIdAndNameAndReadUrl: + return FileIdAndNameAndReadUrl(fileId=file_id, name=name, url=url) + + +def test_fetch_deacon_idx_only_when_raw_reads_enabled(config, monkeypatch): + class _ExitLoop(Exception): # noqa: N818 + """Attaches to `fetch_unprocessed_sequences` to exit run()'s + processing loop after startup. + """ + + fetch_index_mock = MagicMock(return_value=True) + monkeypatch.setattr(prepro, "run_deacon_fetch", fetch_index_mock) + monkeypatch.setattr(prepro, "fetch_unprocessed_sequences", MagicMock(side_effect=_ExitLoop)) + + # RAW_READS not enabled for this organism -> the index must not be fetched. + with pytest.raises(_ExitLoop): + prepro.run(config) + fetch_index_mock.assert_not_called() + + # RAW_READS enabled -> the index is fetched once, into the dataset dir. + config.submission_file_categories = [FileCategory.RAW_READS] + with pytest.raises(_ExitLoop): + prepro.run(config) + fetch_index_mock.assert_called_once() + (index_path,) = fetch_index_mock.call_args.args + assert index_path.endswith(prepro.DEACON_INDEX) + + +def test_process_submitted_files_runs_raw_reads_check(config, tmp_path, monkeypatch): + validate_mock = MagicMock(return_value=([], [])) + monkeypatch.setattr(file_processing_functions, "validate_raw_reads_submission", validate_mock) + + files = [fastq_file("f1", "reads.fastq")] + errors, warnings = process_submitted_files( + config, str(tmp_path), {FileCategory.RAW_READS: files} + ) + + validate_mock.assert_called_once_with(config, str(tmp_path), files, 0.05) + assert errors == [] + assert warnings == [] + + +def test_process_submitted_files_skips_empty_category(config, tmp_path, monkeypatch): + # The backend sends an empty list for file categories without files + monkeypatch.setattr( + file_processing_functions, + "validate_raw_reads_submission", + lambda *a: pytest.fail("empty category must not be validated"), + ) + file_mapping = {FileCategory.RAW_READS: []} + errors, warnings = process_submitted_files(config, str(tmp_path), file_mapping) + assert errors == [] + assert warnings == [] + + +def test_process_submitted_files_reports_unsupported_category(config, tmp_path): + file_mapping = {FileCategory.ANNOTATIONS: [fastq_file("f1", "notes.txt")]} + errors, _ = process_submitted_files(config, str(tmp_path), file_mapping) + assert len(errors) == 1 + assert "not supported by preprocessing" in errors[0].message + + +@pytest.mark.usefixtures("mock_deacon") +def test_single_valid_fastq_passes(config, tmp_path): + files = [fastq_file("f1", "reads.fastq")] + errors, warnings = validate_raw_reads_submission( + config, str(tmp_path), files, MAX_HOST_PROPORTION + ) + assert errors == [] + assert warnings == [] + + +@pytest.mark.usefixtures("mock_deacon") +def test_paired_valid_fastq_passes(config, tmp_path): + files = [fastq_file("f1", "reads_R1.fastq"), fastq_file("f2", "reads_R2.fastq.gz")] + errors, warnings = validate_raw_reads_submission( + config, str(tmp_path), files, MAX_HOST_PROPORTION + ) + assert errors == [] + assert warnings == [] + + +@pytest.mark.usefixtures("mock_deacon") +def test_too_many_files_reports_error(config, tmp_path): + files = [ + fastq_file("f1", "r1.fastq"), + fastq_file("f2", "r2.fastq"), + fastq_file("f3", "r3.fastq"), + ] + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + assert len(errors) == 1 + assert "Received 3" in errors[0].message + + +@pytest.mark.usefixtures("mock_deacon") +def test_unrecognized_extension_reports_error(config, tmp_path): + files = [fastq_file("f1", "reads.txt")] + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + assert len(errors) == 1 + assert "unrecognized extension" in errors[0].message + + +def test_host_contamination_within_threshold(config, tmp_path, mock_deacon): + mock_deacon(0.01) + files = [fastq_file("f1", "reads.fastq")] + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + assert errors == [] + + +def test_host_contamination_exceeds_threshold(config, tmp_path, mock_deacon): + mock_deacon(0.9) + files = [fastq_file("f1", "reads.fastq")] + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + assert len(errors) == 1 + assert "host reads proportion" in errors[0].message + + +def test_missing_url_reports_error_and_skips_deacon(config, tmp_path, monkeypatch): + def fail_if_called(*args, **kwargs): + msg = "deacon must not run when the file URL is missing" + raise AssertionError(msg) + + monkeypatch.setattr(file_processing_functions, "run_deacon_filter", fail_if_called) + files = [fastq_file("f1", "reads.fastq", url=None)] + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + assert len(errors) == 1 + assert "No URL for file" in errors[0].message diff --git a/preprocessing/nextclade/tests/test_nextclade_preprocessing.py b/preprocessing/nextclade/tests/test_nextclade_preprocessing.py index 65ba48726f..e65d3b1578 100644 --- a/preprocessing/nextclade/tests/test_nextclade_preprocessing.py +++ b/preprocessing/nextclade/tests/test_nextclade_preprocessing.py @@ -26,8 +26,6 @@ ) from loculus_preprocessing.datatypes import ( AnnotationSourceType, - FileCategory, - FileIdAndNameAndReadUrl, SegmentClassificationMethod, SubmissionData, UnprocessedData, @@ -286,145 +284,6 @@ def invalid_sequence() -> str: sequenceNameToFastaId={"main": "fastaHeader"}, ), ), - Case( - name="with file", - input_metadata={}, - input_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), - ] - }, - input_sequence={"fastaHeader": sequence_with_mutation("single")}, - accession_id="1", - expected_metadata={ - "completeness": 1.0, - "totalInsertedNucs": 0, - "totalSnps": 1, - "totalDeletedNucs": 0, - "length": len(consensus_sequence("single")), - "nonExistentField": "None", - "variant": True, - }, - expected_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), - ] - }, - expected_errors=[], - expected_warnings=[], - expected_processed_alignment=ProcessedAlignment( - unalignedNucleotideSequences={"main": sequence_with_mutation("single")}, - alignedNucleotideSequences={"main": sequence_with_mutation("single")}, - nucleotideInsertions={}, - alignedAminoAcidSequences={ - "NPEbolaSudan": ebola_sudan_aa(sequence_with_mutation("single"), "NP"), - "VP35EbolaSudan": ebola_sudan_aa(sequence_with_mutation("single"), "VP35"), - }, - aminoAcidInsertions={}, - sequenceNameToFastaId={"main": "fastaHeader"}, - ), - ), - Case( - name="with too many raw read files", - input_metadata={}, - input_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), - FileIdAndNameAndReadUrl(fileId="file-id-0003", name="reads_R3.fastq"), - ] - }, - input_sequence={"fastaHeader": sequence_with_mutation("single")}, - accession_id="1", - expected_metadata={ - "completeness": 1.0, - "totalInsertedNucs": 0, - "totalSnps": 1, - "totalDeletedNucs": 0, - "length": len(consensus_sequence("single")), - "nonExistentField": "None", - "variant": True, - }, - expected_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads_R1.fastq"), - FileIdAndNameAndReadUrl(fileId="file-id-0002", name="reads_R2.fastq"), - FileIdAndNameAndReadUrl(fileId="file-id-0003", name="reads_R3.fastq"), - ] - }, - expected_errors=build_processing_annotations( - [ - ProcessingAnnotationHelper( - ["reads_R1.fastq", "reads_R2.fastq", "reads_R3.fastq"], - ["reads_R1.fastq", "reads_R2.fastq", "reads_R3.fastq"], - "Received 3 for raw reads upload. Please submit raw reads as one or two FASTQ files containing raw single-end or paired-end reads.", - AnnotationSourceType.FILE, - ) - ] - ), - expected_warnings=[], - expected_processed_alignment=ProcessedAlignment( - unalignedNucleotideSequences={"main": sequence_with_mutation("single")}, - alignedNucleotideSequences={"main": sequence_with_mutation("single")}, - nucleotideInsertions={}, - alignedAminoAcidSequences={ - "NPEbolaSudan": ebola_sudan_aa(sequence_with_mutation("single"), "NP"), - "VP35EbolaSudan": ebola_sudan_aa(sequence_with_mutation("single"), "VP35"), - }, - aminoAcidInsertions={}, - sequenceNameToFastaId={"main": "fastaHeader"}, - ), - ), - Case( - name="with unrecognized raw read file extension", - input_metadata={}, - input_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") - ] - }, - input_sequence={"fastaHeader": sequence_with_mutation("single")}, - accession_id="1", - expected_metadata={ - "completeness": 1.0, - "totalInsertedNucs": 0, - "totalSnps": 1, - "totalDeletedNucs": 0, - "length": len(consensus_sequence("single")), - "nonExistentField": "None", - "variant": True, - }, - expected_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") - ] - }, - expected_errors=build_processing_annotations( - [ - ProcessingAnnotationHelper( - ["reads.txt"], - ["reads.txt"], - "Raw reads file 'reads.txt' has unrecognized extension. " - "Allowed extensions: .fastq, .fastq.gz, .fq, .fq.gz", - AnnotationSourceType.FILE, - ) - ] - ), - expected_warnings=[], - expected_processed_alignment=ProcessedAlignment( - unalignedNucleotideSequences={"main": sequence_with_mutation("single")}, - alignedNucleotideSequences={"main": sequence_with_mutation("single")}, - nucleotideInsertions={}, - alignedAminoAcidSequences={ - "NPEbolaSudan": ebola_sudan_aa(sequence_with_mutation("single"), "NP"), - "VP35EbolaSudan": ebola_sudan_aa(sequence_with_mutation("single"), "VP35"), - }, - aminoAcidInsertions={}, - sequenceNameToFastaId={"main": "fastaHeader"}, - ), - ), ] single_segment_failed_case_definitions = [ @@ -1275,55 +1134,6 @@ def invalid_sequence() -> str: sequenceNameToFastaId={"ebola-sudan": "ebola-sudan"}, ), ), - Case( - name="validate raw read files when sequences are not aligned", - input_metadata={}, - input_sequence={ - "prefix_ebola-sudan": sequence_with_mutation("ebola-sudan"), - "other_prefix_ebola-zaire": sequence_with_mutation("ebola-zaire"), - }, - accession_id="1", - input_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") - ] - }, - expected_files={ - FileCategory.RAW_READS: [ - FileIdAndNameAndReadUrl(fileId="file-id-0001", name="reads.txt") - ] - }, - expected_metadata={ - "length_ebola-sudan": len(consensus_sequence("ebola-sudan")), - "length_ebola-zaire": len(consensus_sequence("ebola-zaire")), - }, - expected_errors=build_processing_annotations( - [ - ProcessingAnnotationHelper( - ["reads.txt"], - ["reads.txt"], - "Raw reads file 'reads.txt' has unrecognized extension. " - "Allowed extensions: .fastq, .fastq.gz, .fq, .fq.gz", - AnnotationSourceType.FILE, - ) - ] - ), - expected_warnings=[], - expected_processed_alignment=ProcessedAlignment( - unalignedNucleotideSequences={ - "ebola-sudan": sequence_with_mutation("ebola-sudan"), - "ebola-zaire": sequence_with_mutation("ebola-zaire"), - }, - alignedNucleotideSequences={}, - nucleotideInsertions={}, - alignedAminoAcidSequences={}, - aminoAcidInsertions={}, - sequenceNameToFastaId={ - "ebola-sudan": "prefix_ebola-sudan", - "ebola-zaire": "other_prefix_ebola-zaire", - }, - ), - ), ] From 9f90c0689e460517b6c5d4a609a30e3f2389a918 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Fri, 10 Jul 2026 15:46:59 +0200 Subject: [PATCH 07/13] TO REVERT: update deployment to only have west-nile, raise prepro memory --- kubernetes/loculus/values_preview_server.yaml | 22 ++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/kubernetes/loculus/values_preview_server.yaml b/kubernetes/loculus/values_preview_server.yaml index 7200693acb..7c262b00a6 100644 --- a/kubernetes/loculus/values_preview_server.yaml +++ b/kubernetes/loculus/values_preview_server.yaml @@ -142,7 +142,7 @@ resources: memory: "200Mi" cpu: "10m" limits: - memory: "3Gi" + memory: "8Gi" minio: requests: memory: "250Mi" @@ -156,3 +156,23 @@ defaultResources: limits: memory: "1Gi" cpu: "20m" + +# Preview only: run just the west-nile preprocessing pipeline while testing the deacon +# host-depletion step. Each enabled organism runs its own prepro pod and fetches the +# ~3.1Gi panhuman index, so disabling the rest keeps startup and memory manageable. +# Deep-merges `enabled: false` onto the base defaultOrganisms entries. +defaultOrganisms: + cchf: + enabled: false + cchf-multi-ref: + enabled: false + dummy-organism: + enabled: false + dummy-organism-with-files: + enabled: false + ebola-sudan: + enabled: false + enteroviruses: + enabled: false + not-aligned-organism: + enabled: false From 319db4dc70459c396c3abfd422697183ac1a1572 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Fri, 10 Jul 2026 15:47:53 +0200 Subject: [PATCH 08/13] Log deacon filter stderr, switch to --deplete mode --- .../file_processing_functions.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index f0847de314..00bb1381ab 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -151,11 +151,21 @@ def run_deacon_fetch(index_path: str) -> bool: def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconSummary: - args = ["deacon", "filter", "--threads", "1", "--summary", summary_json, index, input_file] + args = [ + "deacon", + "filter", + "--deplete", + "--threads", + "1", + "--summary", + summary_json, + index, + input_file, + ] logger.debug(f"Running Deacon filter on '{input_file}': {args}") exit_code = subprocess.run( # noqa: S603 - args, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + args, check=False, stdout=subprocess.DEVNULL ).returncode if exit_code != 0: message = f"Deacon filter failed with exit code {exit_code}" From 4618873897dede19a8671fc200b3e48ad422e070 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Mon, 13 Jul 2026 12:24:33 +0200 Subject: [PATCH 09/13] Start deacon server only when needed --- .../file_processing_functions.py | 36 +++++++++++++++++++ .../src/loculus_preprocessing/prepro.py | 13 ++++++- 2 files changed, 48 insertions(+), 1 deletion(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index 00bb1381ab..e4b5c7ad30 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -172,3 +172,39 @@ def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconS logger.error(message) raise Exception(message) return DeaconSummary.from_json(summary_json) + + +def start_deacon_server() -> subprocess.Popen: + args = [ + "deacon", + "server", + "start", + ] + logger.debug("Starting Deacon server") + + return subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # noqa: S603 + + +def stop_deacon_server(proc: subprocess.Popen | None) -> int: + args = ["deacon", "--use-server", "server", "stop"] + logger.debug("Stopping Deacon server") + + exit_code = subprocess.run( # noqa: S603 + args, + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ).returncode + + if proc is None: + return exit_code + + try: + return proc.wait(timeout=5) + except subprocess.TimeoutExpired: # server didn't stop -> kill manually + proc.terminate() + try: + return proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + return proc.wait() diff --git a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py index cceda1478d..5b066de33b 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py @@ -9,6 +9,8 @@ from loculus_preprocessing.file_processing_functions import ( process_submitted_files, run_deacon_fetch, + start_deacon_server, + stop_deacon_server, ) from .backend import ( @@ -724,7 +726,8 @@ def run(config: Config) -> None: # noqa: C901 msg = "Diamond database URL must be provided for diamond segment classification" raise ValueError(msg) download_diamond_db(config, dataset_dir + "/diamond/diamond.dmnd") - if FileCategory.RAW_READS in config.submission_file_categories: + can_submit_raw_reads = FileCategory.RAW_READS in config.submission_file_categories + if can_submit_raw_reads: # TODO: Should we host this ourselves somewhere instead of fetching from external? index_path = os.path.join(dataset_dir, DEACON_INDEX) if not run_deacon_fetch(index_path): @@ -749,6 +752,9 @@ def run(config: Config) -> None: # noqa: C901 # Don't use etag if we just got data # preprocessing only asks for 100 sequences to process at a time, so there might be more etag = None + proc = None + if can_submit_raw_reads: + proc = start_deacon_server() try: processed = process_all(unprocessed, dataset_dir, config) except Exception as e: @@ -756,6 +762,11 @@ def run(config: Config) -> None: # noqa: C901 f"Processing failed. Traceback : {e}. Unprocessed data: {unprocessed}" ) continue + if can_submit_raw_reads: + exit_code = stop_deacon_server(proc) + if exit_code != 0: + message = f"Stopping Deacon server failed with exit code: {exit_code}" + logger.error(message) if config.create_embl_file: upload_flatfiles(processed, config) From f6f1eed9985bf70172dbb7050e66265d8a112639 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Mon, 13 Jul 2026 14:41:55 +0200 Subject: [PATCH 10/13] Some robustness --- .../file_processing_functions.py | 28 ++++++++----------- .../src/loculus_preprocessing/prepro.py | 12 +++----- 2 files changed, 15 insertions(+), 25 deletions(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index e4b5c7ad30..fdcb326575 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -138,7 +138,7 @@ def validate_raw_reads_submission( return errors, warnings -def run_deacon_fetch(index_path: str) -> bool: +def run_deacon_fetch(index_path: str) -> None: args = ["deacon", "index", "fetch", "--output", index_path, "panhuman-1"] logger.debug(f"Running Deacon fetch: {args}") @@ -146,13 +146,13 @@ def run_deacon_fetch(index_path: str) -> bool: if exit_code != 0: message = f"Deacon fetch failed with exit code {exit_code}" logger.error(message) - return False - return True + raise RuntimeError(message) def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconSummary: args = [ "deacon", + "--use-server", "filter", "--deplete", "--threads", @@ -185,26 +185,20 @@ def start_deacon_server() -> subprocess.Popen: return subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # noqa: S603 -def stop_deacon_server(proc: subprocess.Popen | None) -> int: +def stop_deacon_server(proc: subprocess.Popen) -> None: args = ["deacon", "--use-server", "server", "stop"] logger.debug("Stopping Deacon server") - exit_code = subprocess.run( # noqa: S603 + subprocess.run( # noqa: S603 args, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - ).returncode - - if proc is None: - return exit_code + ) try: - return proc.wait(timeout=5) - except subprocess.TimeoutExpired: # server didn't stop -> kill manually - proc.terminate() - try: - return proc.wait(timeout=5) - except subprocess.TimeoutExpired: - proc.kill() - return proc.wait() + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + logger.warning("Failed to stop Deacon server gracefully, sending SIGKILL") + proc.kill() + proc.wait() diff --git a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py index 5b066de33b..4ef92a126c 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py @@ -730,9 +730,7 @@ def run(config: Config) -> None: # noqa: C901 if can_submit_raw_reads: # TODO: Should we host this ourselves somewhere instead of fetching from external? index_path = os.path.join(dataset_dir, DEACON_INDEX) - if not run_deacon_fetch(index_path): - msg = "Failed to fetch Deacon index" - raise RuntimeError(msg) + run_deacon_fetch(index_path) total_processed = 0 etag = None @@ -762,11 +760,9 @@ def run(config: Config) -> None: # noqa: C901 f"Processing failed. Traceback : {e}. Unprocessed data: {unprocessed}" ) continue - if can_submit_raw_reads: - exit_code = stop_deacon_server(proc) - if exit_code != 0: - message = f"Stopping Deacon server failed with exit code: {exit_code}" - logger.error(message) + finally: + if proc is not None: + stop_deacon_server(proc) if config.create_embl_file: upload_flatfiles(processed, config) From b7c1890e7499fb12614978aa41dd1ebddedf1d6e Mon Sep 17 00:00:00 2001 From: maverbiest Date: Mon, 13 Jul 2026 15:11:18 +0200 Subject: [PATCH 11/13] Refactor, download files as stream --- .../nextclade/src/loculus_preprocessing/backend.py | 8 ++++---- .../nextclade/src/loculus_preprocessing/prepro.py | 9 ++++----- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/backend.py b/preprocessing/nextclade/src/loculus_preprocessing/backend.py index f7eea9cedc..ba92bc49c2 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/backend.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/backend.py @@ -248,11 +248,11 @@ def upload_embl_file_to_presigned_url( raise RuntimeError(msg) -# TODO: Stream data instead of loading all at once with write_bytes def download_file(config: Config, url: str, save_path: str) -> None: - response = requests.get(url, timeout=config.backend_request_timeout_seconds) - response.raise_for_status() - Path(save_path).write_bytes(response.content) + with requests.get(url, stream=True, timeout=config.backend_request_timeout_seconds) as response: + response.raise_for_status() + with Path(save_path).open("wb") as f: + f.writelines(response.iter_content(chunk_size=1024 * 1024)) def download_minimizer(config, save_path): diff --git a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py index 4ef92a126c..04d6faf7d4 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py @@ -728,7 +728,6 @@ def run(config: Config) -> None: # noqa: C901 download_diamond_db(config, dataset_dir + "/diamond/diamond.dmnd") can_submit_raw_reads = FileCategory.RAW_READS in config.submission_file_categories if can_submit_raw_reads: - # TODO: Should we host this ourselves somewhere instead of fetching from external? index_path = os.path.join(dataset_dir, DEACON_INDEX) run_deacon_fetch(index_path) @@ -750,9 +749,9 @@ def run(config: Config) -> None: # noqa: C901 # Don't use etag if we just got data # preprocessing only asks for 100 sequences to process at a time, so there might be more etag = None - proc = None + deacon_server_process = None if can_submit_raw_reads: - proc = start_deacon_server() + deacon_server_process = start_deacon_server() try: processed = process_all(unprocessed, dataset_dir, config) except Exception as e: @@ -761,8 +760,8 @@ def run(config: Config) -> None: # noqa: C901 ) continue finally: - if proc is not None: - stop_deacon_server(proc) + if deacon_server_process is not None: + stop_deacon_server(deacon_server_process) if config.create_embl_file: upload_flatfiles(processed, config) From 0e1345de52b079bf356ef1a5b772497908c3cd44 Mon Sep 17 00:00:00 2001 From: maverbiest Date: Wed, 15 Jul 2026 11:40:48 +0200 Subject: [PATCH 12/13] Run deacon on both files at once for paired submission --- .../file_processing_functions.py | 46 ++++++++++++------- .../nextclade/tests/test_file_processing.py | 2 +- 2 files changed, 30 insertions(+), 18 deletions(-) diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index fdcb326575..b0eb69b34f 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -2,6 +2,7 @@ import os import subprocess # noqa: S404 from tempfile import TemporaryDirectory +from uuid import uuid4 from requests import HTTPError @@ -98,9 +99,13 @@ def validate_raw_reads_submission( deacon_index = os.path.join(dataset_dir, DEACON_INDEX) with TemporaryDirectory() as tmp_dir: + local_files = [] for file in files: if (url := file.url) is None: - message = f"No URL for file '{file.name}' with id '{file.fileId}'" + message = ( + f"Internal error: no URL for file '{file.name}' with id " + f"'{file.fileId}'. Please contact the administrator." + ) errors.append( ProcessingAnnotation.from_single( name=file.name, type=AnnotationSourceType.FILE, message=message @@ -120,20 +125,27 @@ def validate_raw_reads_submission( ) ) continue - - summary_json_path = os.path.join(tmp_dir, f"{file.fileId}_summary.json") - deacon_summary = run_deacon_filter(deacon_index, file_name_internal, summary_json_path) - if deacon_summary.seqs_removed_proportion > max_host_proportion: - message = ( - f"File {file.name} (id: {file.fileId}) had a host reads proportion of " - f"{deacon_summary.seqs_removed_proportion}, maximum allowed proportion " - f"is {max_host_proportion}" - ) - errors.append( - ProcessingAnnotation.from_single( - name=file.name, type=AnnotationSourceType.FILE, message=message - ) + local_files.append(file_name_internal) + if errors: + return errors, warnings + + summary_json_path = os.path.join(tmp_dir, f"{uuid4()}_summary.json") + deacon_summary = run_deacon_filter(deacon_index, local_files, summary_json_path) + if deacon_summary.seqs_removed_proportion > max_host_proportion: + names = ", ".join(file.name for file in files) + message = ( + f"File(s) '{names}' had a host reads proportion of " + f"{deacon_summary.seqs_removed_proportion}, maximum allowed proportion " + f"is {max_host_proportion}" + ) + errors.append( + ProcessingAnnotation.from_fields( + input_fields=[f.name for f in files], + output_fields=[f.name for f in files], + type=AnnotationSourceType.FILE, + message=message, ) + ) return errors, warnings @@ -149,7 +161,7 @@ def run_deacon_fetch(index_path: str) -> None: raise RuntimeError(message) -def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconSummary: +def run_deacon_filter(index: str, input_files: list[str], summary_json: str) -> DeaconSummary: args = [ "deacon", "--use-server", @@ -160,9 +172,9 @@ def run_deacon_filter(index: str, input_file: str, summary_json: str) -> DeaconS "--summary", summary_json, index, - input_file, + *input_files, ] - logger.debug(f"Running Deacon filter on '{input_file}': {args}") + logger.debug(f"Running Deacon filter on '{', '.join(input_files)}': {args}") exit_code = subprocess.run( # noqa: S603 args, check=False, stdout=subprocess.DEVNULL diff --git a/preprocessing/nextclade/tests/test_file_processing.py b/preprocessing/nextclade/tests/test_file_processing.py index bba2808fef..e20a542a43 100644 --- a/preprocessing/nextclade/tests/test_file_processing.py +++ b/preprocessing/nextclade/tests/test_file_processing.py @@ -186,4 +186,4 @@ def fail_if_called(*args, **kwargs): files = [fastq_file("f1", "reads.fastq", url=None)] errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) assert len(errors) == 1 - assert "No URL for file" in errors[0].message + assert "no URL for file" in errors[0].message From f468c6d91705529014550234324895954a256a9b Mon Sep 17 00:00:00 2001 From: maverbiest Date: Wed, 15 Jul 2026 13:52:01 +0200 Subject: [PATCH 13/13] Make host contamination threshold configurable --- kubernetes/loculus/values.schema.json | 8 ++++ kubernetes/loculus/values.yaml | 1 + .../src/loculus_preprocessing/config.py | 10 +++++ .../file_processing_functions.py | 8 ++-- .../src/loculus_preprocessing/prepro.py | 2 +- .../nextclade/tests/test_file_processing.py | 43 ++++++++++--------- 6 files changed, 47 insertions(+), 25 deletions(-) diff --git a/kubernetes/loculus/values.schema.json b/kubernetes/loculus/values.schema.json index 134655edc6..aeac087ed5 100644 --- a/kubernetes/loculus/values.schema.json +++ b/kubernetes/loculus/values.schema.json @@ -814,6 +814,14 @@ "type": "string", "description": "Minimizer used for nextclade sort (if require_nextclade_sort_match or if segments are to be assigned using nextclade sort)" }, + "deacon_max_host_proportion": { + "groups": ["nextcladePipelineConfigFile"], + "docsIncludePrefix": false, + "type": "number", + "minimum": 0, + "maximum": 1, + "description": "Maximum proportion of host reads in submitted raw read files, as detected by Deacon." + }, "backend_request_timeout_seconds": { "groups": ["nextcladePipelineConfigFile"], "docsIncludePrefix": false, diff --git a/kubernetes/loculus/values.yaml b/kubernetes/loculus/values.yaml index 96a6c84f31..552c3bdb98 100644 --- a/kubernetes/loculus/values.yaml +++ b/kubernetes/loculus/values.yaml @@ -1558,6 +1558,7 @@ defaultOrganismConfig: &defaultOrganismConfig - name: singleReference genes: [] nextclade_dataset_server: https://data.clades.nextstrain.org/v3 + deacon_max_host_proportion: 0.01 ingest: &ingest image: ghcr.io/loculus-project/ingest configFile: diff --git a/preprocessing/nextclade/src/loculus_preprocessing/config.py b/preprocessing/nextclade/src/loculus_preprocessing/config.py index e4bdf4b4f3..e0f890e39e 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/config.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/config.py @@ -10,6 +10,7 @@ from pydantic import BaseModel, Field, model_validator from loculus_preprocessing.datatypes import ( + FileCategory, FunctionArgs, FunctionInputs, FunctionName, @@ -120,6 +121,7 @@ class Config(BaseModel): require_nextclade_sort_match: bool = False minimizer_url: str | None = None diamond_dmnd_url: str | None = None + deacon_max_host_proportion: float | None = None submission_file_categories: list[str] = Field(default_factory=list) create_embl_file: bool = False @@ -144,6 +146,14 @@ def finalize(self): if not self.backend_host: # Set here so we can use organism self.backend_host = f"http://127.0.0.1:8079/{self.organism}" + if FileCategory.RAW_READS in self.submission_file_categories: + if self.deacon_max_host_proportion is None: + msg = "deacon_max_host_proportion must be set if raw reads submissions are enabled" + raise ValueError(msg) + if not (0.0 <= self.deacon_max_host_proportion <= 1.0): + msg = "deacon_max_host_proportion must be in [0.0, 1.0]" + raise ValueError(msg) + self.processing_order = get_processing_order(self) return self diff --git a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py index b0eb69b34f..dba2938ff2 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/file_processing_functions.py @@ -2,6 +2,7 @@ import os import subprocess # noqa: S404 from tempfile import TemporaryDirectory +from typing import cast from uuid import uuid4 from requests import HTTPError @@ -37,7 +38,6 @@ def process_submitted_files( config, dataset_dir, files, - 0.05, # TODO: make configurable ) errors.extend(rr_errors) warnings.extend(rr_warnings) @@ -61,7 +61,6 @@ def validate_raw_reads_submission( config: Config, dataset_dir: str, files: list[FileIdAndNameAndReadUrl], - max_host_proportion: float, ) -> tuple[list[ProcessingAnnotation], list[ProcessingAnnotation]]: errors: list[ProcessingAnnotation] = [] warnings: list[ProcessingAnnotation] = [] @@ -131,12 +130,13 @@ def validate_raw_reads_submission( summary_json_path = os.path.join(tmp_dir, f"{uuid4()}_summary.json") deacon_summary = run_deacon_filter(deacon_index, local_files, summary_json_path) - if deacon_summary.seqs_removed_proportion > max_host_proportion: + # deacon_max_host_proportion is guaranteed not None by Config.finalize() + if deacon_summary.seqs_removed_proportion > cast(float, config.deacon_max_host_proportion): names = ", ".join(file.name for file in files) message = ( f"File(s) '{names}' had a host reads proportion of " f"{deacon_summary.seqs_removed_proportion}, maximum allowed proportion " - f"is {max_host_proportion}" + f"is {config.deacon_max_host_proportion}" ) errors.append( ProcessingAnnotation.from_fields( diff --git a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py index 04d6faf7d4..10a6a77108 100644 --- a/preprocessing/nextclade/src/loculus_preprocessing/prepro.py +++ b/preprocessing/nextclade/src/loculus_preprocessing/prepro.py @@ -712,7 +712,7 @@ def upload_flatfiles(processed: Sequence[SubmissionData], config: Config) -> Non ) -def run(config: Config) -> None: # noqa: C901 +def run(config: Config) -> None: # noqa: C901, PLR0912 with TemporaryDirectory(delete=not config.keep_tmp_dir) as dataset_dir: if config.alignment_requirement != AlignmentRequirement.NONE: download_nextclade_dataset(dataset_dir, config) diff --git a/preprocessing/nextclade/tests/test_file_processing.py b/preprocessing/nextclade/tests/test_file_processing.py index e20a542a43..99a27b5c06 100644 --- a/preprocessing/nextclade/tests/test_file_processing.py +++ b/preprocessing/nextclade/tests/test_file_processing.py @@ -21,7 +21,10 @@ @pytest.fixture def config() -> Config: - return Config() + return Config( + submission_file_categories=[FileCategory.RAW_READS], + deacon_max_host_proportion=MAX_HOST_PROPORTION, + ) @pytest.fixture @@ -74,19 +77,23 @@ class _ExitLoop(Exception): # noqa: N818 monkeypatch.setattr(prepro, "run_deacon_fetch", fetch_index_mock) monkeypatch.setattr(prepro, "fetch_unprocessed_sequences", MagicMock(side_effect=_ExitLoop)) - # RAW_READS not enabled for this organism -> the index must not be fetched. - with pytest.raises(_ExitLoop): - prepro.run(config) - fetch_index_mock.assert_not_called() - - # RAW_READS enabled -> the index is fetched once, into the dataset dir. - config.submission_file_categories = [FileCategory.RAW_READS] + # RAW_READS enabled -> the index is fetched once with pytest.raises(_ExitLoop): prepro.run(config) fetch_index_mock.assert_called_once() (index_path,) = fetch_index_mock.call_args.args assert index_path.endswith(prepro.DEACON_INDEX) + fetch_index_mock = MagicMock(return_value=True) + monkeypatch.setattr(prepro, "run_deacon_fetch", fetch_index_mock) + + # RAW_READS not enabled -> the index must not be fetched. + config.submission_file_categories = [] + config.deacon_max_host_proportion = None + with pytest.raises(_ExitLoop): + prepro.run(config) + fetch_index_mock.assert_not_called() + def test_process_submitted_files_runs_raw_reads_check(config, tmp_path, monkeypatch): validate_mock = MagicMock(return_value=([], [])) @@ -97,7 +104,7 @@ def test_process_submitted_files_runs_raw_reads_check(config, tmp_path, monkeypa config, str(tmp_path), {FileCategory.RAW_READS: files} ) - validate_mock.assert_called_once_with(config, str(tmp_path), files, 0.05) + validate_mock.assert_called_once_with(config, str(tmp_path), files) assert errors == [] assert warnings == [] @@ -125,9 +132,7 @@ def test_process_submitted_files_reports_unsupported_category(config, tmp_path): @pytest.mark.usefixtures("mock_deacon") def test_single_valid_fastq_passes(config, tmp_path): files = [fastq_file("f1", "reads.fastq")] - errors, warnings = validate_raw_reads_submission( - config, str(tmp_path), files, MAX_HOST_PROPORTION - ) + errors, warnings = validate_raw_reads_submission(config, str(tmp_path), files) assert errors == [] assert warnings == [] @@ -135,9 +140,7 @@ def test_single_valid_fastq_passes(config, tmp_path): @pytest.mark.usefixtures("mock_deacon") def test_paired_valid_fastq_passes(config, tmp_path): files = [fastq_file("f1", "reads_R1.fastq"), fastq_file("f2", "reads_R2.fastq.gz")] - errors, warnings = validate_raw_reads_submission( - config, str(tmp_path), files, MAX_HOST_PROPORTION - ) + errors, warnings = validate_raw_reads_submission(config, str(tmp_path), files) assert errors == [] assert warnings == [] @@ -149,7 +152,7 @@ def test_too_many_files_reports_error(config, tmp_path): fastq_file("f2", "r2.fastq"), fastq_file("f3", "r3.fastq"), ] - errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files) assert len(errors) == 1 assert "Received 3" in errors[0].message @@ -157,7 +160,7 @@ def test_too_many_files_reports_error(config, tmp_path): @pytest.mark.usefixtures("mock_deacon") def test_unrecognized_extension_reports_error(config, tmp_path): files = [fastq_file("f1", "reads.txt")] - errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files) assert len(errors) == 1 assert "unrecognized extension" in errors[0].message @@ -165,14 +168,14 @@ def test_unrecognized_extension_reports_error(config, tmp_path): def test_host_contamination_within_threshold(config, tmp_path, mock_deacon): mock_deacon(0.01) files = [fastq_file("f1", "reads.fastq")] - errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files) assert errors == [] def test_host_contamination_exceeds_threshold(config, tmp_path, mock_deacon): mock_deacon(0.9) files = [fastq_file("f1", "reads.fastq")] - errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files) assert len(errors) == 1 assert "host reads proportion" in errors[0].message @@ -184,6 +187,6 @@ def fail_if_called(*args, **kwargs): monkeypatch.setattr(file_processing_functions, "run_deacon_filter", fail_if_called) files = [fastq_file("f1", "reads.fastq", url=None)] - errors, _ = validate_raw_reads_submission(config, str(tmp_path), files, MAX_HOST_PROPORTION) + errors, _ = validate_raw_reads_submission(config, str(tmp_path), files) assert len(errors) == 1 assert "no URL for file" in errors[0].message