Rename dataset loader and add compression argument to more components

src/datasets/api/comp_dataset_loader.yaml

@@ -12,4 +12,9 @@ arguments:
     __merge__: file_raw.yaml
     direction: "output"
     required: true
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
 test_resources: []

src/datasets/api/comp_normalization.yaml

@@ -16,6 +16,11 @@ arguments:
     __merge__: file_normalized.yaml
     direction: output
     required: true
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
   - name: "--normalization_id"
     type: string
     description: "The normalization id to store in the dataset metadata. If not specified, the functionality name will be used."

src/datasets/api/comp_processor_hvg.yaml

@@ -20,6 +20,11 @@ arguments:
     direction: output
     __merge__: file_hvg.yaml
     required: true
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
   - name: "--var_hvg"
     type: string
     default: "hvg"

src/datasets/api/comp_processor_knn.yaml

@@ -20,6 +20,11 @@ arguments:
     direction: output
     __merge__: file_knn.yaml
     required: true
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
   - name: "--key_added"
     type: string
     default: "knn"

src/datasets/api/comp_processor_pca.yaml

@@ -24,6 +24,11 @@ arguments:
     direction: output
     __merge__: file_pca.yaml
     required: true
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
   - name: "--obsm_embedding"
     type: string
     default: "X_pca"

src/datasets/api/comp_processor_subset.yaml

@@ -22,6 +22,11 @@ arguments:
     __merge__: file_common_dataset.yaml
     direction: output
     required: false
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
 test_resources:
   - path: /resources_test/common/pancreas
     dest: resources_test/common/pancreas

src/datasets/api/comp_processor_svd.yaml

@@ -28,6 +28,11 @@ arguments:
     direction: output
     __merge__: file_svd.yaml
     required: false
+  - name: --output_compression
+    type: string
+    choices: [gzip, lzf]
+    required: false
+    example: gzip
   - name: "--obsm_embedding"
     type: string
     default: "X_svd"

src/datasets/loaders/scrnaseq/op3/config.vsh.yaml

@@ -1,4 +1,4 @@
-name: op3
+name: openproblems_op3
 namespace: datasets/loaders/scrnaseq
 description: |
   "Loads and preprocesses the OP3 dataset from GEO accession GSE279945."
@@ -39,7 +39,7 @@ argument_groups:
       - name: "--dataset_id"
         type: string
         description: "Unique identifier for the dataset"
-        default: "op3"
+        default: "openproblems_op3"
       - name: "--dataset_name"
         type: string
         description: "Human-readable name for the dataset"

src/datasets/processors/hvg/script.py

@@ -32,5 +32,4 @@
 adata.var[par["var_hvg_score"]] = out['dispersions_norm'].values
 
 print(">> Writing data", flush=True)
-adata.write_h5ad(par['output'])
-
+adata.write_h5ad(par['output'], compression=par["output_compression"])

src/datasets/processors/knn/script.py

@@ -23,5 +23,5 @@
 )
 
 print(">> Writing data", flush=True)
-adata.write_h5ad(par['output'])
+adata.write_h5ad(par['output'], compression=par["output_compression"])
 

src/datasets/processors/pca/script.py

@@ -35,5 +35,4 @@
 }
 
 print(">> Writing data", flush=True)
-adata.write_h5ad(par['output'])
-
+adata.write_h5ad(par['output'], compression=par["output_compression"])

src/datasets/processors/subsample/script.py

@@ -140,6 +140,6 @@
     del adata_output_mod2.X
 
 print(">> Writing data", flush=True)
-adata_output.write_h5ad(par["output"])
+adata_output.write_h5ad(par["output"], compression=par["output_compression"])
 if par["output_mod2"] is not None:
-    adata_output_mod2.write_h5ad(par["output_mod2"])
+    adata_output_mod2.write_h5ad(par["output_mod2"], compression=par["output_compression"])

src/datasets/processors/svd/script.py

@@ -39,7 +39,7 @@
 
 
 print(">> Writing data", flush=True)
-adata.write_h5ad(par["output"])
+adata.write_h5ad(par["output"], compression=par["output_compression"])
 if par["input_mod2"] is not None:
-    adata2.write_h5ad(par["output_mod2"])
+    adata2.write_h5ad(par["output_mod2"], compression=par["output_compression"])
 

src/datasets/resource_scripts/op3.sh → src/datasets/resource_scripts/openproblems_op3.sh

@@ -6,7 +6,7 @@ params_file="/tmp/datasets_op3.yaml"
 
 cat > "$params_file" << 'HERE'
 param_list:
-  - id: op3
+  - id: openproblems_op3
     input: https://ftp.ncbi.nlm.nih.gov/geo/series/GSE279nnn/GSE279945/suppl/GSE279945_sc_counts_processed.h5ad
     dataset_name: "OP3: single-cell multimodal dataset in PBMCs for perturbation prediction benchmarking"
     dataset_summary: "The Open Problems Perurbation Prediction (OP3) dataset with small molecule perturbations in PBMCs"
@@ -24,9 +24,9 @@ output_normalized: force_null
 output_pca: force_null
 output_hvg: force_null
 output_knn: force_null
-publish_dir: s3://openproblems-data/resources/datasets/op3
+publish_dir: s3://openproblems-data/resources/datasets
 HERE
-
+<
 cat > "/tmp/nextflow.config" << 'HERE'
 process {
   withName:'.*publishStatesProc' {
@@ -36,19 +36,20 @@ process {
 }
 HERE
 
-tw launch https://github.com/openproblems-bio/openproblems.git \
- --revision main_build \
- --pull-latest \
- --main-script target/nextflow/datasets/workflows/scrnaseq/process_op3/main.nf \
- --workspace 53907369739130 \
- --params-file "$params_file" \
- --labels op3,dataset_loader \
- --config /tmp/nextflow.config
+# tw launch https://github.com/openproblems-bio/openproblems.git \
+#  --revision main_build \
+#  --pull-latest \
+#  --main-script target/nextflow/datasets/workflows/scrnaseq/process_openproblems_op3/main.nf \
+#  --workspace 53907369739130 \
+#  --params-file "$params_file" \
+#  --labels op3,dataset_loader \
+#  --config /tmp/nextflow.config
 
-# set -x
-# nextflow run . \
-#   -main-script target/nextflow/datasets/workflows/scrnaseq/process_op3/main.nf \
-#   -profile docker \
-#   -resume \
-#   -params-file "$params_file" \
-#   -config /tmp/nextflow.config
+set -x
+export AWS_PROFILE=op
+nextflow run . \
+  -main-script target/nextflow/datasets/workflows/scrnaseq/process_openproblems_op3/main.nf \
+  -profile docker \
+  -resume \
+  -params-file "$params_file" \
+  -config /tmp/nextflow.config

src/datasets/resource_scripts/op3_test.sh → src/datasets/resource_scripts/openproblems_op3_test.sh

@@ -19,11 +19,11 @@ HERE
 
 set -x
 nextflow run . \
-  -main-script target/nextflow/datasets/workflows/scrnaseq/process_op3/main.nf \
+  -main-script target/nextflow/datasets/workflows/scrnaseq/process_openproblems_op3/main.nf \
   -profile docker \
   -resume \
   --input https://ftp.ncbi.nlm.nih.gov/geo/series/GSE279nnn/GSE279945/suppl/GSE279945_sc_counts_processed.h5ad \
-  --id op3 \
+  --id openproblems_sop3 \
   --dataset_name "OP3: single-cell multimodal dataset in PBMCs for perturbation prediction benchmarking" \
   --dataset_summary "The Open Problems Perurbation Prediction (OP3) dataset with small molecule perturbations in PBMCs" \
   --dataset_description "The OP3 dataset is to-date the largest single-cell small molecule perturbation dataset in primary tissue with multiple donor replicates." \

src/datasets/workflows/scrnaseq/process_op3/config.vsh.yaml

@@ -1,4 +1,4 @@
-name: process_op3
+name: process_openproblems_op3
 namespace: datasets/workflows/scrnaseq
 description: |
   "Fetch, filter, normalize, and prepare datasets from the Open Problems Perturbation Prediction (OP3) dataset 
@@ -37,7 +37,7 @@ argument_groups:
         type: string
         description: "Unique identifier."
         required: true
-        default: op3
+        default: openproblems_op3
       - name: --dataset_name
         type: string
         description: "Nicely formatted name."
@@ -115,6 +115,11 @@ argument_groups:
         __merge__: /src/datasets/api/file_common_dataset.yaml
         direction: output
         required: true
+      - name: --output_compression
+        type: string
+        choices: [gzip, lzf]
+        required: false
+        example: gzip
       - name: --output_meta
         type: file
         direction: output
@@ -147,7 +152,7 @@ resources:
     entrypoint: run_wf
   - path: /common/nextflow_helpers/helper.nf
 dependencies:
-  - name: datasets/loaders/scrnaseq/op3
+  - name: datasets/loaders/scrnaseq/openproblems_op3
   - name: datasets/normalization/log_cp
   - name: datasets/normalization/log_scran_pooling
   - name: datasets/normalization/sqrt_cp

src/datasets/workflows/scrnaseq/process_op3/main.nf

@@ -49,7 +49,7 @@ workflow run_wf {
     }
 
     // fetch data from OP3 dataset
-    | op3.run(
+    | openproblems_op3.run(
       fromState: [
 	"input": "input",
         "data_type": "data_type",
@@ -115,7 +115,7 @@ workflow run_wf {
     )
 
     | knn.run(
-      fromState: ["input": "output_pca"],
+      fromState: ["input": "output_pca", "output_compression": "output_compression"],
       toState: ["output_knn": "output"]
     )