Merge remote-tracking branch 'origin/Feature/coassembly-schema' into feature/coassembly-schema

Author: SandyRogers

deployment/ebi-wp-k8s-hl/Taskfile.yaml

@@ -26,7 +26,7 @@ tasks:
     cmds:
       - kubectl --context=emgapiv2-hl-exp exec -it $(task get-api-pod-name) -- python manage.py prefectcli deploy {{ .FILE | default (list "workflows/flows/" .FLOW ".py" | join "") }}:{{.FLOW}} --name {{.FLOW}}_deployment -p {{.POOL}} --prefect-file workflows/prefect_deployments/prefect-ebi-codon.yaml
     vars:
-      POOL: "slurm_2"
+      POOL: "slurm"
     requires:
       vars: [FLOW]
 

deployment/ebi-wp-k8s-hl/start_prefect_slurm_worker.sh

@@ -9,6 +9,6 @@ source /nfs/production/rdf/metagenomics/jenkins-slurm/dev-prefect-agent/deployme
 source /nfs/production/rdf/metagenomics/jenkins-slurm/dev-prefect-agent/venv/bin/activate
 
 cd /nfs/production/rdf/metagenomics/jenkins-slurm/dev-prefect-agent/
-python manage.py prefectcli worker start --type "slurm" --pool "slurm_2" >> /nfs/production/rdf/metagenomics/jenkins-slurm/logs/dev-prefect-slurm-worker.log 2>&1 &
+python manage.py prefectcli worker start --type "slurm" --pool "slurm" >> /nfs/production/rdf/metagenomics/jenkins-slurm/logs/dev-prefect-slurm-worker.log 2>&1 &
 
 echo $! > "${DJANGOPIDFILE}"

workflows/flows/analyse_study_tasks/raw_reads/run_rawreads_pipeline_via_samplesheet.py

@@ -47,15 +47,15 @@ def run_rawreads_pipeline_via_samplesheet(
 ):
     if workdir is None:
         workdir = (
-            Path(f"{EMG_CONFIG.slurm.default_nextflow_workdir}")
-            / Path(f"{mgnify_study.ena_study.accession}")
-            / f"{EMG_CONFIG.rawreads_pipeline.pipeline_name}_{EMG_CONFIG.amplicon_pipeline.pipeline_version}"
+            Path(EMG_CONFIG.slurm.default_nextflow_workdir)
+            / mgnify_study.ena_study.accession
+            / f"{EMG_CONFIG.rawreads_pipeline.pipeline_name}_{EMG_CONFIG.rawreads_pipeline.pipeline_version}"
         )
     if outdir is None:
         outdir = (
-            Path(f"{EMG_CONFIG.slurm.default_workdir}")
-            / Path(f"{mgnify_study.ena_study.accession}")
-            / f"{EMG_CONFIG.rawreads_pipeline.pipeline_name}_{EMG_CONFIG.amplicon_pipeline.pipeline_version}"
+            Path(EMG_CONFIG.slurm.default_workdir)
+            / mgnify_study.ena_study.accession
+            / f"{EMG_CONFIG.rawreads_pipeline.pipeline_name}_{EMG_CONFIG.rawreads_pipeline.pipeline_version}"
         )
 
     rawreads_analyses = analyses.models.Analysis.objects.select_related("run").filter(

workflows/flows/analysis/assembly/flows/run_assembly_analysis_pipeline_batch.py

@@ -142,6 +142,14 @@ def run_assembly_analysis_pipeline_batch(
         )
         nextflow_workdir.mkdir(parents=True, exist_ok=True)
 
+        nextflow_workdir = (
+            Path(EMG_CONFIG.slurm.default_nextflow_workdir)
+            / mgnify_study.ena_study.accession
+            / f"{EMG_CONFIG.assembly_analysis_pipeline.pipeline_name}_{EMG_CONFIG.assembly_analysis_pipeline.pipeline_version}"
+            / f"{assembly_analysis_batch.id}"
+        )
+        nextflow_workdir.mkdir(parents=True, exist_ok=True)
+
         command = cli_command(
             [
                 (
@@ -163,12 +171,6 @@ def run_assembly_analysis_pipeline_batch(
                     EMG_CONFIG.assembly_analysis_pipeline.pipeline_nf_profile,
                 ),
                 "-resume",
-                (
-                    "-work-dir",
-                    Path(EMG_CONFIG.assembly_analysis_pipeline.workdir_root)
-                    / mgnify_study.first_accession
-                    / "asa",
-                ),
                 EMG_CONFIG.assembly_analysis_pipeline.has_fire_access
                 and "--use_fire_download",
                 ("--input", samplesheet),

workflows/flows/analysis/assembly/flows/run_map_batch.py

@@ -2,7 +2,6 @@
 from datetime import timedelta
 from pathlib import Path
 
-from django.utils.text import slugify
 from django.db import close_old_connections
 from django.db.models import Q
 from prefect import flow, get_run_logger
@@ -151,9 +150,10 @@ def run_map_batch(assembly_analyses_batch_id: uuid.UUID):
     logger.info(f"Using output dir {map_outdir} for MAP pipeline")
 
     nextflow_workdir = (
-        Path(assembly_analysis_batch.workspace_dir)
-        / "map"
-        / f"map-sheet-{slugify(map_samplesheet_path)}"
+        Path(EMG_CONFIG.slurm.default_nextflow_workdir)
+        / mgnify_study.ena_study.accession
+        / f"{EMG_CONFIG.map_pipeline.pipeline_name}_{EMG_CONFIG.map_pipeline.pipeline_version}"
+        / f"{assembly_analysis_batch.id}"
     )
     nextflow_workdir.mkdir(parents=True, exist_ok=True)
 

workflows/flows/analysis/assembly/flows/run_virify_batch.py

@@ -2,7 +2,6 @@
 from datetime import timedelta
 from pathlib import Path
 
-from django.utils.text import slugify
 from django.db import close_old_connections
 from prefect import flow, get_run_logger
 from prefect.runtime import flow_run
@@ -137,9 +136,10 @@ def run_virify_batch(assembly_analyses_batch_id: uuid.UUID):
     logger.info(f"Using output dir {virify_outdir} for VIRify pipeline")
 
     nextflow_workdir = (
-        Path(assembly_analysis_batch.workspace_dir)
-        / "virify"
-        / f"virify-sheet-{slugify(virify_samplesheet_path)}"
+        Path(EMG_CONFIG.slurm.default_nextflow_workdir)
+        / mgnify_study.ena_study.accession
+        / f"{EMG_CONFIG.virify_pipeline.pipeline_name}_{EMG_CONFIG.virify_pipeline.pipeline_version}"
+        / f"{assembly_analysis_batch.id}"
     )
     nextflow_workdir.mkdir(parents=True, exist_ok=True)
 

workflows/prefect_deployments/prefect-ebi-codon.yaml

@@ -36,7 +36,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/analysis_assembly_study.py:analysis_assembly_study
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: run_assembly_analysis_pipeline_batch_deployment
   description: |-
@@ -60,7 +60,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/run_assembly_analysis_pipeline_batch.py:run_assembly_analysis_pipeline_batch
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: run_virify_batch_deployment
   description: |-
@@ -82,7 +82,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/run_virify_batch.py:run_virify_batch
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: run_map_batch_deployment
   description: |-
@@ -104,7 +104,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/run_map_batch.py:run_map_batch
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: import_asa_batch_deployment
   description: |-
@@ -117,7 +117,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/import_asa_batch.py:import_asa_batch
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: import_virify_batch_deployment
   description: |-
@@ -127,7 +127,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/import_virify_batch.py:import_virify_batch
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: import_map_batch_deployment
   description: |-
@@ -137,7 +137,7 @@ deployments:
   entrypoint: workflows/flows/analysis/assembly/flows/import_map_batch.py:import_map_batch
   schedule: null
   work_pool:
-    name: slurm_2
+    name: slurm
 
 - name: finalize_assembly_study_deployment
   description: |-
@@ -159,4 +159,21 @@ deployments:
     :type study_accession: str
   entrypoint: workflows/flows/analysis/assembly/flows/finalize_assembly_study.py:finalize_assembly_study
   work_pool:
-    name: slurm_2
+    name: slurm
+
+- name: nextflow_trace_etl_flow
+  description: |-
+    Nextflow Trace extraction and transformation flow.
+
+    This flow orchestrates the extraction and transformation of Nextflow trace data
+    from OrchestratedClusterJob models.
+
+    :param sqlite_db_path: Path to the SQLite database where the transformed data will be stored
+    :param batch_size: Number of database records to process at once
+    :param min_created_at: Only process jobs created after this datetime
+    :param max_created_at: Only process jobs created before this datetime
+    :param only_completed: Only process completed jobs
+    :param exclude_failed: Exclude failed jobs
+  entrypoint: workflows/flows/nf_traces/flows.py:nextflow_trace_etl_flow
+  work_pool:
+    name: slurm