Fix stuff that failed in full run

.gitignore

@@ -10,6 +10,6 @@ trace-*
 .ipynb_checkpoints
 /temp
 /.vscode
-
+*_iterative_test.*
 /resources_raw
 /*.tar

scripts/run_benchmark/run_full_seqeracloud.sh

@@ -17,16 +17,16 @@ cat > /tmp/params.yaml << HERE
 input_states: s3://openproblems-data/resources/task_cyto_batch_integration/datasets/**/state.yaml
 rename_keys: 'input_censored_split1:output_censored_split1;input_censored_split2:output_censored_split2;input_unintegrated:output_unintegrated'
 output_state: "state.yaml"
-# settings: '{"methods_include": ["combat", "gaussnorm"]}'
+settings: '{"metrics_exclude": ["cms"], "methods_include": ["mnnpy", "cytovi"]}'
 publish_dir: "$publish_dir"
 HERE
 
 tw launch https://github.com/openproblems-bio/task_cyto_batch_integration.git \
-  --revision build/main \
+  --revision build/fix_failed_stuff \
   --pull-latest \
   --main-script target/nextflow/workflows/run_benchmark/main.nf \
   --workspace 53907369739130 \
   --params-file /tmp/params.yaml \
   --entry-name auto \
   --config common/nextflow_helpers/labels_tw.config \
-  --labels task_cyto_batch_integration,full
+  --labels task_cyto_batch_integration,mnnnpy

src/control_methods/perfect_integration/script.py

@@ -4,8 +4,8 @@
 # The following code has been auto-generated by Viash.
 par = {
     "input_unintegrated": "resources_test/task_cyto_batch_integration/mouse_spleen_flow_cytometry_subset/unintegrated.h5ad",
-    "output_integrated_split1": "resources_test/task_cyto_batch_integration/mouse_spleen_flow_cytometry_subset/integrated_split1.h5ad",
-    "output_integrated_split2": "resources_test/task_cyto_batch_integration/mouse_spleen_flow_cytometry_subset/integrated_split2.h5ad",
+    "output_integrated_split1": "resources_test/task_cyto_batch_integration/mouse_spleen_flow_cytometry_subset/perfect_integrated_split1.h5ad",
+    "output_integrated_split2": "resources_test/task_cyto_batch_integration/mouse_spleen_flow_cytometry_subset/perfect_integrated_split2.h5ad",
 }
 meta = {"name": "perfect_integration"}
 

src/control_methods/shuffle_integration_by_batch/config.vsh.yaml

@@ -3,7 +3,7 @@ name: shuffle_integration_by_batch
 label: Shuffle integration by batch
 summary: Integrations are randomly permuted within each batch
 description: Integrations are randomly permuted within each batch
-status: disabled
+# status: disabled
 resources:
   - type: python_script
     path: script.py

src/control_methods/shuffle_integration_by_batch/script.py

@@ -16,26 +16,47 @@
 
 print("Reading and preparing input files", flush=True)
 adata = ad.read_h5ad(par["input_unintegrated"])
+adata_split1 = adata[(adata.obs.is_control > 0) | (adata.obs.batch == 1)].copy()
+adata_split2 = adata[(adata.obs.is_control > 0) | (adata.obs.split == 2)].copy()
 
-adata.obs["batch_str"] = adata.obs["batch"].astype(str)
+print("Randomise features - split 1", flush=True)
+adata_split1.obs["batch_str"] = adata_split1.obs["batch"].astype(str)
+integrated = _randomize_features(
+    adata_split1.layers["preprocessed"],
+    partition=adata_split1.obs["batch"],
+)
+
+# create new anndata
+output_split1 = ad.AnnData(
+    obs=adata_split1.obs[[]],
+    var=adata_split1.var[[]],
+    layers={"integrated": integrated},
+    uns={
+        "dataset_id": adata_split1.uns["dataset_id"],
+        "method_id": meta["name"],
+        "parameters": {},
+    },
+)
 
-print("Randomise features", flush=True)
+print("Randomise features - split 2", flush=True)
+adata_split2.obs["batch_str"] = adata_split2.obs["batch"].astype(str)
 integrated = _randomize_features(
-    adata.layers["preprocessed"],
-    partition=adata.obs["batch"],
+    adata_split2.layers["preprocessed"],
+    partition=adata_split2.obs["batch"],
 )
 
 # create new anndata
-output = ad.AnnData(
-    obs=adata.obs[[]],
-    var=adata.var[[]],
+output_split2 = ad.AnnData(
+    obs=adata_split2.obs[[]],
+    var=adata_split2.var[[]],
     layers={"integrated": integrated},
     uns={
-        "dataset_id": adata.uns["dataset_id"],
+        "dataset_id": adata_split2.uns["dataset_id"],
         "method_id": meta["name"],
         "parameters": {},
     },
 )
 
 print("Write output AnnData to file", flush=True)
-output.write_h5ad(par["output"], compression="gzip")
+output_split1.write_h5ad(par["output_integrated_split1"], compression="gzip")
+output_split2.write_h5ad(par["output_integrated_split2"], compression="gzip")

src/control_methods/shuffle_integration_by_cell_type/script.py

@@ -16,28 +16,47 @@
 
 print("Reading and preparing input files", flush=True)
 adata = ad.read_h5ad(par["input_unintegrated"])
+adata_split1 = adata[(adata.obs.is_control > 0) | (adata.obs.batch == 1)].copy()
+adata_split2 = adata[(adata.obs.is_control > 0) | (adata.obs.split == 2)].copy()
 
-adata.obs["batch_str"] = adata.obs["batch"].astype(str)
-
-print("Randomise features", flush=True)
+print("Randomise features - split 1", flush=True)
+adata_split1.obs["batch_str"] = adata_split1.obs["batch"].astype(str)
 integrated = _randomize_features(
-    adata.layers["preprocessed"],
-    partition=adata.obs["cell_type"],
+    adata_split1.layers["preprocessed"],
+    partition=adata_split1.obs["cell_type"],
 )
 
 # create new anndata
-output = ad.AnnData(
-    obs=adata.obs[[]],
-    var=adata.var[[]],
+output_split1 = ad.AnnData(
+    obs=adata_split1.obs[[]],
+    var=adata_split1.var[[]],
     layers={"integrated": integrated},
     uns={
-        "dataset_id": adata.uns["dataset_id"],
+        "dataset_id": adata_split1.uns["dataset_id"],
         "method_id": meta["name"],
         "parameters": {},
     },
 )
 
-all(x==y for x,y in zip(output.var_names, adata.var_names))
+print("Randomise features - split 2", flush=True)
+adata_split2.obs["batch_str"] = adata_split2.obs["batch"].astype(str)
+integrated = _randomize_features(
+    adata_split2.layers["preprocessed"],
+    partition=adata_split2.obs["cell_type"],
+)
+
+# create new anndata
+output_split2 = ad.AnnData(
+    obs=adata_split2.obs[[]],
+    var=adata_split2.var[[]],
+    layers={"integrated": integrated},
+    uns={
+        "dataset_id": adata_split2.uns["dataset_id"],
+        "method_id": meta["name"],
+        "parameters": {},
+    },
+)
 
 print("Write output AnnData to file", flush=True)
-output.write_h5ad(par["output"], compression="gzip")
+output_split1.write_h5ad(par["output_integrated_split1"], compression="gzip")
+output_split2.write_h5ad(par["output_integrated_split2"], compression="gzip")

src/methods/batchadjust_all_controls/config.vsh.yaml

@@ -66,6 +66,7 @@ resources:
     path: script.R
   - path: /src/utils/anndata_to_fcs.R
   - path: BatchAdjust.R
+  - path: utils.R
 
 engines:
   # Specifications for the Docker image for this component.

src/methods/batchadjust_all_controls/script.R

@@ -2,15 +2,19 @@ library(anndata)
 library(flowCore)
 ## VIASH START
 par <- list(
-  input = "resources_test/.../input.h5ad",
-  output = "output.h5ad"
+  input = "resources_test/debug/batchadjust/_viash_par/input_1/censored_split1.h5ad",
+  output = "resources_test/debug/batchadjust/output.h5ad",
+  percentile = as.integer('80')
 )
 meta <- list(
   name = "batchadjust_all_controls",
-  temp_dir = "/tmp"
+  temp_dir = "/tmp",
+  resources_dir = "src/methods/batchadjust_all_controls"
 )
+source("src/utils/anndata_to_fcs.R")
 ## VIASH END
 
+source(paste0(meta$resources_dir, "/utils.R"))
 source(paste0(meta$resources_dir, "/anndata_to_fcs.R"))
 source(paste0(meta$resources_dir, "/BatchAdjust.R"))
 
@@ -33,7 +37,11 @@ source(paste0(meta$resources_dir, "/BatchAdjust.R"))
 cat("Reading input files\n")
 input <- anndata::read_h5ad(par[["input"]])
 #use Original_ID column to restore cell order after I/O operations
-input$layers["preprocessed"][, "Original_ID"] <- seq(1, dim(input)[1])
+# input$layers["preprocessed"][, "Original_ID"] <- seq(1, dim(input)[1])
+
+original_id_in_var <- "Original_ID" %in% input$var_names
+
+input <- add_original_id(input)
 
 cat("Split cells\n")
 input_controls <- input[input$obs$is_control != 0, ]
@@ -48,9 +56,14 @@ print(input_no_controls)
 
 #avoid NA due to invalid factor level
 input_controls$obs$sample <- as.character(input_controls$obs$sample)
+
+
+# make sure there is _ after the batch1 or batch2, otherwise batchadjust won't find the fcs files.
+input_no_controls$obs$sample <- sapply(input_no_controls$obs$sample, fix_batch_underscore_anynum)
+
 # Set sample names for batch-specific control files
-input_controls$obs$sample[input_controls$obs$batch == 1] <- "0Batch1_anchor"
-input_controls$obs$sample[input_controls$obs$batch == 2] <- "0Batch2_anchor"
+input_controls$obs$sample[input_controls$obs$batch == 1] <- "Batch1_anchor"
+input_controls$obs$sample[input_controls$obs$batch == 2] <- "Batch2_anchor"
 
 cat("Writing FCS files\n")
 anndata_to_fcs(input_controls, out_dir = meta[["temp_dir"]])
@@ -72,7 +85,7 @@ BatchAdjust(
   outdir = output_dir,
   channelsFile = paste0(meta[["temp_dir"]], "/to_correct_list.txt"),
   anchorKeyword = "anchor",
-  batchKeyword = "atch", #skip 'b' to make it robust to upper/lowercase
+  batchKeyword = "Batch", #skip 'b' to make it robust to upper/lowercase
   method = perc,
   transformation = FALSE,
   addExt = NULL,
@@ -96,6 +109,11 @@ if (FALSE %in% order_check) {
   stop("Failed in restoring indexing")
 }
 
+# Remove Original_ID if it was not there in the beginning
+if (!original_id_in_var) {
+    corrected_matrix$Original_ID <- NULL
+}
+
 cat("Write output AnnData to file\n")
 output <- anndata::AnnData(
   obs = input$obs[, integer(0)],

src/methods/batchadjust_all_controls/utils.R

@@ -0,0 +1,51 @@
+fix_batch_underscore_anynum <- function(x) {
+    # Pattern: Batch followed by one or more digits, NOT followed by underscore
+    pattern <- "(Batch\\d+)(?!_)"
+
+    # Add underscore if missing
+    x <- sub(pattern, "\\1_", x, perl = TRUE)
+
+    return(x)
+}
+
+add_original_id <- function(input) {
+    if (!"Original_ID" %in% input$var_names) {
+        cat("Adding Original_ID to var and recreating input anndata\n")
+        old_var <- input$var
+        old_var[] <- lapply(old_var, function(x) if (is.factor(x)) as.character(x) else x)
+        old_var <- rbind(old_var,
+            data.frame(
+                numeric_id=length(input$var_names) + 1,
+                channel="Original_ID",
+                marker="",
+                marker_type="other",
+                to_correct=FALSE,
+                row.names = "Original_ID"
+            )
+        )
+        old_var[] <- lapply(old_var, function(x) if (is.character(x)) as.factor(x) else x)
+        new_mat <- Matrix::as.matrix(input$layers[["preprocessed"]])
+        new_mat <- cbind(new_mat, Original_ID = seq_len(nrow(new_mat)))
+
+        # create new anndata
+        input <- anndata::AnnData(
+            X = new_mat,
+            obs = input$obs,
+            var = old_var,
+            layers = list(preprocessed = new_mat),
+            uns = list(
+                dataset_description = input$uns$dataset_description,
+                dataset_id = input$uns$dataset_id,
+                dataset_name = input$uns$dataset_name,
+                dataset_organism = input$uns$dataset_organism,
+                dataset_reference = input$uns$dataset_reference,
+                dataset_summary = input$uns$dataset_summary,
+                dataset_url = input$uns$dataset_url
+            )
+        )
+    } else {
+        cat("Adding new Original_ID var\n")
+        input$layers["preprocessed"][, "Original_ID"] <- seq(1, dim(input)[1])
+    }
+    return(input)
+}

src/methods/batchadjust_one_control/config.vsh.yaml

@@ -66,6 +66,7 @@ resources:
     path: script.R
   - path: /src/utils/anndata_to_fcs.R
   - path: BatchAdjust.R
+  - path: utils.R
 
 engines:
   # Specifications for the Docker image for this component.

src/methods/batchadjust_one_control/script.R

@@ -2,15 +2,18 @@ library(anndata)
 library(flowCore)
 ## VIASH START
 par <- list(
-  input = "resources_test/.../input.h5ad",
-  output = "output.h5ad"
+  input = "resources_test/debug/batchadjust/_viash_par/input_1/censored_split1.h5ad",
+  output = "resources_test/debug/batchadjust/output.h5ad",
+  percentile = as.integer('80')
 )
 meta <- list(
   name = "batchadjust_all_controls",
-  temp_dir = "/tmp"
+  temp_dir = "resources_test/tmp",
+  resources_dir = 
 )
 ## VIASH END
 
+source(paste0(meta$resources_dir, "/utils.R"))
 source(paste0(meta$resources_dir, "/anndata_to_fcs.R"))
 source(paste0(meta$resources_dir, "/BatchAdjust.R"))
 
@@ -33,7 +36,11 @@ source(paste0(meta$resources_dir, "/BatchAdjust.R"))
 cat("Reading input files\n")
 input <- anndata::read_h5ad(par[["input"]])
 #use Original_ID column to restore cell order after I/O operations
-input$layers["preprocessed"][, "Original_ID"] <- seq(1, dim(input)[1])
+# input$layers["preprocessed"][, "Original_ID"] <- seq(1, dim(input)[1])
+
+original_id_in_var <- "Original_ID" %in% input$var_names
+
+input <- add_original_id(input)
 
 cat("Split cells\n")
 input_controls <- input[input$obs$is_control == 1, ]
@@ -48,9 +55,13 @@ print(input_no_controls)
 
 #avoid NA due to invalid factor level
 input_controls$obs$sample <- as.character(input_controls$obs$sample)
+
+# make sure there is _ after the batch1 or batch2, otherwise batchadjust won't find the fcs files.
+input_no_controls$obs$sample <- sapply(input_no_controls$obs$sample, fix_batch_underscore_anynum)
+
 # Set sample names for batch-specific control files
-input_controls$obs$sample[input_controls$obs$batch == 1] <- "0Batch1_anchor"
-input_controls$obs$sample[input_controls$obs$batch == 2] <- "0Batch2_anchor"
+input_controls$obs$sample[input_controls$obs$batch == 1] <- "Batch1_anchor"
+input_controls$obs$sample[input_controls$obs$batch == 2] <- "Batch2_anchor"
 
 cat("Writing FCS files\n")
 anndata_to_fcs(input_controls, out_dir = meta[["temp_dir"]])
@@ -72,7 +83,7 @@ BatchAdjust(
   outdir = output_dir,
   channelsFile = paste0(meta[["temp_dir"]], "/to_correct_list.txt"),
   anchorKeyword = "anchor",
-  batchKeyword = "atch", #skip 'b' to make it robust to upper/lowercase
+  batchKeyword = "Batch", #skip 'b' to make it robust to upper/lowercase
   method = perc,
   transformation = FALSE,
   addExt = NULL,
@@ -96,6 +107,11 @@ if (FALSE %in% order_check) {
   stop("Failed in restoring indexing")
 }
 
+# Remove Original_ID if it was not there in the beginning
+if (!original_id_in_var) {
+    corrected_matrix$Original_ID <- NULL
+}
+
 cat("Write output AnnData to file\n")
 output <- anndata::AnnData(
   obs = input$obs[, integer(0)],