73e2eea4be
GET_FINAL_METABOLITES_STATIC took chembl_db as a `path` input, so Nextflow copied the ~9.2 GB SQLite DB into each task's work directory. With ~29 concurrent tasks that is hundreds of GB of I/O, against a process that also declared only 1 GB of memory. Every task failed with exit 1 and retried ten times, producing 746 errors and an empty 1b_final_metabolites/ output. Pass the DB as a `val` path instead so tasks read it in place from the dreamdock-data PVC, and give the process parameterised growing memory.
181 lines
8.9 KiB
Plaintext
181 lines
8.9 KiB
Plaintext
nextflow.enable.dsl=2
|
|
|
|
// params.container_biotransformer = 'harbor.cluster.omic.ai/omic/digitaltrials/biotransformer:latest'
|
|
// params.container_conplex = 'harbor.cluster.omic.ai/omic/digitaltrials/conplex_dig_pat:latest'
|
|
// params.container_tissue = 'harbor.cluster.omic.ai/omic/digitaltrials/tissue:latest'
|
|
// params.container_preprocess = 'harbor.cluster.omic.ai/omic/metabolite-screen:project'
|
|
// params.container_mass_screen = 'harbor.cluster.omic.ai/omic/metabolite-screen:adaptive'
|
|
// params.container_chembl = 'harbor.cluster.omic.ai/omic/digitaltrials/chembl:1.0.0'
|
|
|
|
params.container_biotransformer = 'harbor.cluster.omic.ai/omic/digitaltrials/biotransformer@sha256:fb8bdc0b65376bc154b6051ae07079dd9a0e25c3f4c02de73c502002a94d69d5'
|
|
params.container_conplex = 'harbor.cluster.omic.ai/omic/digitaltrials/conplex_dig_pat@sha256:7a3523dba6fa01e3adc9cb79af5e1dcbd2a19d9f92e37cd10df462766078ede3'
|
|
params.container_tissue = 'harbor.cluster.omic.ai/omic/digitaltrials/tissue:1.1.2'
|
|
params.container_preprocess = 'harbor.cluster.omic.ai/omic/metabolite-screen@sha256:872c395e21abd4afea4185b269a8218da4737bd7adbb2cbe2bdf9a1b9c70db17'
|
|
params.container_mass_screen = 'harbor.cluster.omic.ai/omic/metabolite-screen:adaptive-1.2.0'
|
|
params.container_chembl = 'harbor.cluster.omic.ai/omic/digitaltrials/chembl:1.0.0'
|
|
|
|
params.containerOptions = '--rm' // '--gpus all --rm -v /mnt:/mnt'
|
|
|
|
|
|
//BIOTRANSFORMER
|
|
params.project_name = 'test'
|
|
//params.ligands = '/Workspace/next/registry/pipelines/digital_trials/input'
|
|
|
|
//IMPOTRANT!!! use HUMAN. Only one that returns what is expected. metabolism to the last step. It's only one properly set up
|
|
params.mode = 'HUMAN' // Options: SUPER, HUMAN, MASS, ORDERED # only HUMAN is fully implemented
|
|
params.bt_initial_memory = 5 // GB - starting memory for biotransformer
|
|
params.bt_growth_memory = 15 // GB - additional memory per retry
|
|
params.bt_max_retries = 10
|
|
params.bt_fail_action = 'ignore' // 'terminate' or 'ignore'
|
|
params.bt_max_forks = 0 // 0 = unlimited, set to N to limit concurrency
|
|
|
|
// GET_FINAL_METABOLITES_STATIC — queries the ~9.2 GB ChEMBL SQLite DB on the PVC.
|
|
// The DB is opened read-only/immutable and is NOT staged into the task workdir, so the
|
|
// memory below covers rdkit + the biotransformer CSV, not the database itself.
|
|
params.chembl_initial_memory = 4 // GB - starting memory
|
|
params.chembl_growth_memory = 4 // GB - additional memory per retry attempt
|
|
params.chembl_max_retries = 3
|
|
params.chembl_fail_action = 'ignore' // 'terminate' or 'ignore' after retries exhausted
|
|
|
|
//CONPLEX
|
|
params.keep_enst = 'false' //'true' //'false' //to keep individual protein data created by conplex step
|
|
params.conplex_initial_memory = 5 // GB - starting memory for conplex
|
|
params.conplex_growth_memory = 15 // GB - additional memory per retry
|
|
params.conplex_max_retries = 1
|
|
params.conplex_fail_action = 'ignore' // 'terminate' or 'ignore'
|
|
// params.mutated_protein_csv = '/Workspace/next/registry/pipelines/digital_trials/MANE_all_transcipts.csv'
|
|
params.threshold = 0.65 //0.65 //0.8 //0.5 //0.7
|
|
// params.screening_batch_size =100000 //100k is ideal to optimize performance with virtual screening
|
|
params.protein_network_threshold = 0.65 //0.65 //threshold for stirng input
|
|
// NETWORK_ENRICHMENT (string-db) — Python-level urllib3 Retry handles 429/5xx blips first
|
|
// so most rate-limit failures are absorbed without spawning a Nextflow retry workdir.
|
|
params.string_max_forks = 5 // 0 = unlimited; cap below string-db's ~10 req/s throttle
|
|
params.string_max_retries = 1 // Nextflow-level fallback retries (kept low — Python retries are the primary defense)
|
|
params.string_fail_action = 'ignore' // 'terminate' or 'ignore' after max retries exhausted
|
|
params.string_initial_memory = 1 // GB - starting memory; the script reads only transcipt+conplex_score columns
|
|
params.string_growth_memory = 2 // GB - additional memory per retry attempt
|
|
|
|
// TISSUE_DISTRIBUTION + BIO_METRICS (tissue container) — both scripts now use
|
|
// usecols= on significant_interactions.tsv so a 5+ GB input only loads the
|
|
// 'transcipt' (and 'conplex_score' for BIO_METRICS) column.
|
|
params.tissue_initial_memory = 5 // GB
|
|
params.tissue_growth_memory = 5 // GB per retry attempt
|
|
params.tissue_max_retries = 2
|
|
params.tissue_fail_action = 'ignore'
|
|
|
|
params.bio_initial_memory = 5 // GB
|
|
params.bio_growth_memory = 5 // GB per retry attempt
|
|
params.bio_max_retries = 2
|
|
params.bio_fail_action = 'ignore'
|
|
|
|
//TISSUE DISTRIBUTION / BIO PROP
|
|
|
|
// ========================================= ALL RELEVANT FILEPATHS IN THIS SECTION ===================================================
|
|
// Defaults below are the PVC mount paths used for local/k8s execution. WES overrides
|
|
// ligands/outdir via experiment_params (it translates s3:// URIs to PVC mount paths).
|
|
|
|
params.outdir = '/omic/eureka/digital-trial/output'
|
|
|
|
// Ligand CSVs staged in the eureka workspace (s3://omic/eureka/digital-trial/input/ligands/)
|
|
params.ligands = '/omic/eureka/digital-trial/input/ligands'
|
|
|
|
// Intentionally empty: no *.fasta here means PREPROCESS_PROTEIN is never scheduled and
|
|
// the pipeline screens against the prebuilt protein_zarr below. Do NOT point this at a
|
|
// directory containing .fasta files — params.container_preprocess is pinned to a digest
|
|
// that no longer exists in Harbor, so the process would fail on image pull.
|
|
params.mutated_protein_fasta = '/mnt/dreamdock-data/digital_trials/input/blank'
|
|
|
|
// Prebuilt reference data on the dreamdock-data PVC (verified present on k8s-node23)
|
|
params.protein_zarr = '/mnt/dreamdock-data/digital_trials/zarr/protein_seq.zarr' // 7.5 MB
|
|
params.chembl_db = '/mnt/dreamdock-data/digital_trials/chembl/chembl_36.db' // 9.2 GB
|
|
|
|
// =====================================================================================================================================
|
|
|
|
//BIOTRANSFORMER
|
|
include { SUPER_TRANSFORMER } from './main_biotransformer.nf'
|
|
include { HUMAN_TRANSFORMER } from './main_biotransformer.nf'
|
|
include { METABOLITES_BY_MASS } from './main_biotransformer.nf'
|
|
include { ORDERED_SEQUENCE } from './main_biotransformer.nf'
|
|
// include { GET_FINAL_METABOLITES } from './main_biotransformer.nf'
|
|
include { GET_FINAL_METABOLITES_STATIC } from './main_biotransformer.nf'
|
|
//CONPLEX
|
|
include { CONPLEX as CONPLEX_ALL } from './main_conplex.nf'
|
|
include { PREPROCESS_PROTEIN } from './main_conplex.nf'
|
|
|
|
include { MERGE_DRUG } from './main_conplex.nf'
|
|
include { MERGE_INTERACTIONS } from './main_conplex.nf'
|
|
|
|
include { NETWORK_ENRICHMENT } from './main_conplex.nf'
|
|
//TISSUE_DISTRIBUTION / BIO_PROP
|
|
include { TISSUE_DISTRIBUTION } from './main_tissue.nf'
|
|
include { BIO_METRICS } from './main_tissue.nf'
|
|
|
|
workflow {
|
|
//BIOTRANSFORMER
|
|
lig_ch = Channel.fromPath("${params.ligands}/*.csv")
|
|
switch (params.mode) {
|
|
case 'SUPER':
|
|
SUPER_TRANSFORMER(lig_ch)
|
|
break
|
|
case 'HUMAN':
|
|
HUMAN_TRANSFORMER(lig_ch)
|
|
break
|
|
case 'MASS':
|
|
METABOLITES_BY_MASS(lig_ch)
|
|
break
|
|
case 'ORDERED':
|
|
ORDERED_SEQUENCE(lig_ch)
|
|
break
|
|
default:
|
|
println("Invalid mode specified: ${params.mode}")
|
|
}
|
|
// Pass the ChEMBL DB as a plain path string rather than a staged file: at ~9.2 GB,
|
|
// staging it per task copied the DB into every work directory. The process reads it
|
|
// read-only from the PVC mount instead.
|
|
chembl_ch = Channel.value(params.chembl_db)
|
|
// GET_FINAL_METABOLITES(HUMAN_TRANSFORMER.out) // TODO: LOCALIZE
|
|
GET_FINAL_METABOLITES_STATIC(HUMAN_TRANSFORMER.out, chembl_ch)
|
|
//CONPLEX
|
|
|
|
protein_fasta = Channel
|
|
.fromPath("${params.mutated_protein_fasta}/*.fasta")
|
|
.collect()
|
|
// protein_csv = file(params.mutated_protein_csv)
|
|
protein_zarr = PREPROCESS_PROTEIN(protein_fasta).collect().map { it[0] }
|
|
pre_protein_zarr = Channel.fromPath(params.protein_zarr)
|
|
|
|
// Merge both zarr channels into a single list
|
|
merged_zarr = protein_zarr.concat(pre_protein_zarr).collect()
|
|
|
|
lig_with_id = lig_ch.map { csv ->
|
|
[csv.simpleName, csv]
|
|
}
|
|
|
|
metabolite_with_id = GET_FINAL_METABOLITES_STATIC.out.map { txt ->
|
|
def name = txt.simpleName.replaceAll(/_out$/, '')
|
|
[name, txt]
|
|
}
|
|
|
|
// Join on the identifier and remove it
|
|
matched_ch = lig_with_id
|
|
.join(metabolite_with_id)
|
|
// .map { id, ligand, metabolite -> [ligand, metabolite] }
|
|
|
|
// DEBUG_MATCH(matched_ch, merged_zarr)
|
|
|
|
CONPLEX_ALL(matched_ch, merged_zarr)
|
|
|
|
conplex_drug_ch = CONPLEX_ALL.out.map { id, drug, interactions -> drug }
|
|
conplex_interactions_ch = CONPLEX_ALL.out.map { id, drug, interactions -> interactions }
|
|
|
|
NETWORK_ENRICHMENT(conplex_interactions_ch) // TODO: Localize this (use human only)
|
|
//TISSUE DISTRIBUTION
|
|
TISSUE_DISTRIBUTION(conplex_interactions_ch)
|
|
|
|
biometric_ch = matched_ch.join(CONPLEX_ALL.out)
|
|
|
|
// BIO_METRICS(GET_FINAL_METABOLITES.out,conplex_interactions_ch,conplex_drug_ch,lig_ch)
|
|
BIO_METRICS(biometric_ch)
|
|
}
|
|
|