From efa66dc579261b602875fcd898030c0da8be08dd Mon Sep 17 00:00:00 2001 From: Bernd Bohmeier Date: Wed, 25 Mar 2026 16:04:22 +0100 Subject: [PATCH 1/3] Add option to specify number of threads for pipeline The number of threads will be used for mapping and variant calling with delve. --- src/nomadic/map/mappers.py | 14 +++++++------- src/nomadic/process/commands.py | 10 ++++++++++ src/nomadic/realtime/commands.py | 10 ++++++++++ src/nomadic/realtime/factory.py | 3 +++ src/nomadic/realtime/main.py | 10 +++++++++- src/nomadic/realtime/pipelines/barcode.py | 12 +++++++++--- src/nomadic/realtime/steps.py | 7 ++++++- 7 files changed, 54 insertions(+), 12 deletions(-) diff --git a/src/nomadic/map/mappers.py b/src/nomadic/map/mappers.py index 94c6491..83cd67d 100644 --- a/src/nomadic/map/mappers.py +++ b/src/nomadic/map/mappers.py @@ -53,19 +53,19 @@ def map_from_fastqs(self, fastq_dir=None, fastq_paths=None): raise ValueError("Must set either `fastq_dir` or `fastq_paths`.") @abstractmethod - def _define_mapping_command(self, output_bam, flags): + def _define_mapping_command(self, output_bam, *, threads: int, flags): """ Define the command for the mapping algorithm """ pass - def run(self, output_bam, verbose=False): + def run(self, output_bam: str, threads: int, verbose=False): """ Run the mapping algorithm inputs """ # Define the mapping command - self._define_mapping_command(output_bam) + self._define_mapping_command(output_bam, threads=threads) if self.map_cmd is None: raise ValueError("Must define mapping command before running algorithm.") @@ -91,12 +91,12 @@ class Minimap2(MappingAlgorithm): """ - def _define_mapping_command(self, output_bam, flags="--eqx --MD"): + def _define_mapping_command(self, output_bam, *, threads: int, flags="--eqx --MD"): """ Run minimap2, compress result to .bam file, and sort """ - self.map_cmd = "minimap2" + self.map_cmd = f"minimap2 -t {threads}" self.map_cmd += f" -ax map-ont {flags} {shlex.quote(self.reference.fasta_path)} {encode_input_files(self.input_fastqs)} |" self.map_cmd += " samtools view -S -b - |" self.map_cmd += f" samtools sort -o {shlex.quote(output_bam)}" @@ -119,12 +119,12 @@ def create_reference_index(self): index_cmd = f"bwa index {shlex.quote(self.reference.fasta_path)}" subprocess.run(index_cmd, shell=True, check=True) - def _define_mapping_command(self, output_bam, flags=""): + def _define_mapping_command(self, output_bam, *, threads: int, flags=""): """ Run bwa, compress result to .bam file, and sort """ - self.map_cmd = "bwa mem" + self.map_cmd = f"bwa mem -t {threads}" self.map_cmd += " -R '@RG\\tID:misc\\tSM:pool'" # ID and SM tags needed for gatk HaplotypeCaller self.map_cmd += f" {flags} {shlex.quote(self.reference.fasta_path)} {encode_input_files(self.input_fastqs)} |" self.map_cmd += " samtools view -S -b - |" diff --git a/src/nomadic/process/commands.py b/src/nomadic/process/commands.py index 33a5ecf..98515e9 100644 --- a/src/nomadic/process/commands.py +++ b/src/nomadic/process/commands.py @@ -89,6 +89,14 @@ default=False, help="Resume processing a previous experiment if the output directory already exists. This is necessary to pick of processing of an experiment that was aborted.", ) +@click.option( + "-t", + "--threads", + type=int, + default=4, + show_default=True, + help="Number of threads to use for analysis. Note that using more threads can increase the computational load and might lead to slower performance if the computer is not powerful enough.", +) @click.option( "-v", "--verbose", @@ -108,6 +116,7 @@ def process( caller: str, overwrite: bool, resume: bool, + threads: int, verbose: bool, ): """ @@ -166,6 +175,7 @@ def process( region_bed, reference_name, caller, + threads, verbose, with_dashboard=False, host="", diff --git a/src/nomadic/realtime/commands.py b/src/nomadic/realtime/commands.py index d1d989d..03fdaa4 100644 --- a/src/nomadic/realtime/commands.py +++ b/src/nomadic/realtime/commands.py @@ -102,6 +102,14 @@ default=True, help="Whether to start the web dashboard to monitor the run.", ) +@click.option( + "-t", + "--threads", + type=int, + default=4, + show_default=True, + help="Number of threads to use for analysis. Note that using more threads can increase the computational load and might lead to slower performance if the computer is not powerful enough.", +) @click.option( "-v", "--verbose", @@ -136,6 +144,7 @@ def realtime( resume: bool, dashboard: bool, verbose: bool, + threads: int, host: str, port: Optional[int], ): @@ -207,6 +216,7 @@ def realtime( region_bed, reference_name, caller, + threads, verbose, with_dashboard=dashboard, realtime=True, diff --git a/src/nomadic/realtime/factory.py b/src/nomadic/realtime/factory.py index c9d18af..8c37f28 100644 --- a/src/nomadic/realtime/factory.py +++ b/src/nomadic/realtime/factory.py @@ -38,6 +38,7 @@ def __init__( expt_dirs: ExperimentDirectories, fastq_dir: str, caller: str, + threads: int, ref_name: str = "Pf3D7", ): """ @@ -59,6 +60,7 @@ def __init__( self.reference = REFERENCE_COLLECTION[ref_name] self.caller = caller + self.threads = threads def _get_barcode_pipeline(self, barcode_name: str) -> BarcodePipelineRT: """ @@ -70,6 +72,7 @@ def _get_barcode_pipeline(self, barcode_name: str) -> BarcodePipelineRT: "expt_dirs": self.expt_dirs, "bed_path": self.regions.path, "ref_name": self.ref_name, + "threads": self.threads, } if self.caller: diff --git a/src/nomadic/realtime/main.py b/src/nomadic/realtime/main.py index 9b8cde2..821efb9 100644 --- a/src/nomadic/realtime/main.py +++ b/src/nomadic/realtime/main.py @@ -28,6 +28,7 @@ def main( region_bed: str, reference_name: str, caller: str, + threads: int, verbose: bool, host: str, with_dashboard: bool = True, @@ -96,7 +97,14 @@ def main( # INITIALISE WATCHERS factory = PipelineFactory( - expt_name, metadata, regions, expt_dirs, fastq_dir, caller, reference_name + expt_name, + metadata, + regions, + expt_dirs, + fastq_dir, + caller, + ref_name=reference_name, + threads=threads, ) watchers = factory.get_watchers() diff --git a/src/nomadic/realtime/pipelines/barcode.py b/src/nomadic/realtime/pipelines/barcode.py index 8a9f5ff..c4b2e79 100644 --- a/src/nomadic/realtime/pipelines/barcode.py +++ b/src/nomadic/realtime/pipelines/barcode.py @@ -154,6 +154,7 @@ def __init__( expt_dirs: ExperimentDirectories, bed_path: str, caller: str, + threads: int, ref_name: str = "Pf3D7", ): """ @@ -166,7 +167,7 @@ def __init__( # Initialise analysis steps common = {"barcode_name": barcode_name, "expt_dirs": expt_dirs} self.fastq_step = FASTQProcessedRT(**common) - self.map_step = MappingRT(**common, ref_name=ref_name) + self.map_step = MappingRT(**common, ref_name=ref_name, threads=threads) self.flagstat_step = FlagstatsRT(**common, ref_name=ref_name) self.bedcov_step = RegionCoverage( **common, bed_path=bed_path, ref_name=ref_name @@ -176,11 +177,16 @@ def __init__( ) if caller == "delve": self.call_step = CallVariantsRTDelve( - **common, bed_path=bed_path, ref_name=ref_name + **common, + bed_path=bed_path, + ref_name=ref_name, + threads=threads, ) elif caller == "bcftools": self.call_step = CallVariantsRTBcftools( - **common, bed_path=bed_path, ref_name=ref_name + **common, + bed_path=bed_path, + ref_name=ref_name, ) else: raise RuntimeError(f"Unknown caller: {caller}") diff --git a/src/nomadic/realtime/steps.py b/src/nomadic/realtime/steps.py index 38fe9d6..8a6551b 100644 --- a/src/nomadic/realtime/steps.py +++ b/src/nomadic/realtime/steps.py @@ -139,6 +139,7 @@ def __init__( self, barcode_name: str, expt_dirs: ExperimentDirectories, + threads: int, ref_name: str = "Pf3D7", ): """ @@ -153,6 +154,7 @@ def __init__( self.reference = REFERENCE_COLLECTION[ref_name] self.mapper = Minimap2(self.reference) + self.threads = threads self.output_bam = ( f"{self.step_dir}/{self.barcode_name}.{self.reference.name}.final.bam" @@ -181,7 +183,7 @@ def run(self, new_fastqs: List[str], incr_id: str): incr_bam = self._get_incremental_bam_path(incr_id) self.mapper.map_from_fastqs(fastq_paths=new_fastqs) - self.mapper.run(output_bam=incr_bam) + self.mapper.run(output_bam=incr_bam, threads=self.threads) samtools_index(incr_bam) return incr_bam @@ -462,6 +464,7 @@ def __init__( barcode_name: str, expt_dirs: ExperimentDirectories, bed_path: str, + threads: int, ref_name: str = "Pf3D7", ): """Initialise output directory and define file names""" @@ -470,6 +473,7 @@ def __init__( self.bed_path = bed_path self.reference = REFERENCE_COLLECTION[ref_name] + self.threads = threads self.output_dir = produce_dir(self.barcode_dir, self.step_name) self.output_vcf = ( @@ -531,6 +535,7 @@ def run(self, input_bam: str) -> str: f" -f {shlex.quote(self.reference.fasta_path)}" " --set-failed-GTs ." f" {shlex.quote(filtered_bam_path)}" + f" --threads {self.threads}" ) cmd_lowcomplexity_filter = self._get_lowcomplexity_filter_command() From 44705d5a32bd9a28381e18eb1cf16af057d4b6d5 Mon Sep 17 00:00:00 2001 From: Bernd Bohmeier Date: Thu, 18 Jun 2026 14:32:48 +0200 Subject: [PATCH 2/3] Update delve to 0.3.0 This version is needed to specify the number of threads --- build/conda/meta.yaml | 2 +- environments/dev.yml | 2 +- environments/run.yml | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/build/conda/meta.yaml b/build/conda/meta.yaml index 375aa61..2a750ee 100644 --- a/build/conda/meta.yaml +++ b/build/conda/meta.yaml @@ -41,7 +41,7 @@ requirements: - samtools >=1.20 - bcftools >=1.20 - bedtools - - delve-bio =0.2.* + - delve-bio =0.3.* - rsync test: diff --git a/environments/dev.yml b/environments/dev.yml index 8177025..f3c67d1 100644 --- a/environments/dev.yml +++ b/environments/dev.yml @@ -9,7 +9,7 @@ dependencies: - bcftools >=1.20 - htslib - bedtools - - delve-bio =0.2.* + - delve-bio =0.3.* - pysam - jupyter - dash diff --git a/environments/run.yml b/environments/run.yml index 63556d6..33a2763 100644 --- a/environments/run.yml +++ b/environments/run.yml @@ -7,7 +7,7 @@ dependencies: - minimap2 - samtools >=1.20 - bcftools >=1.20 - - delve-bio =0.2.* + - delve-bio =0.3.* - htslib - bedtools - dash From 11c63d6637a1db900864dbb171ad110cc7dff71c Mon Sep 17 00:00:00 2001 From: Bernd Bohmeier Date: Mon, 22 Jun 2026 12:05:34 +0200 Subject: [PATCH 3/3] Set default threads to 5 For our MVP panel we have 10 amplicons, so this allows the variant calling to be parallized in a way that it should do roughly 2 rounds. This also leaves some room if people run two sequencing experiments at the same time. Maybe we should instead check the number of cores in the future. --- src/nomadic/process/commands.py | 2 +- src/nomadic/realtime/commands.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/nomadic/process/commands.py b/src/nomadic/process/commands.py index 98515e9..a7965b7 100644 --- a/src/nomadic/process/commands.py +++ b/src/nomadic/process/commands.py @@ -93,7 +93,7 @@ "-t", "--threads", type=int, - default=4, + default=5, show_default=True, help="Number of threads to use for analysis. Note that using more threads can increase the computational load and might lead to slower performance if the computer is not powerful enough.", ) diff --git a/src/nomadic/realtime/commands.py b/src/nomadic/realtime/commands.py index 03fdaa4..bb0b2cc 100644 --- a/src/nomadic/realtime/commands.py +++ b/src/nomadic/realtime/commands.py @@ -106,7 +106,7 @@ "-t", "--threads", type=int, - default=4, + default=5, show_default=True, help="Number of threads to use for analysis. Note that using more threads can increase the computational load and might lead to slower performance if the computer is not powerful enough.", )