update

R/create_edges_parallel.R

@@ -1,31 +1,55 @@
 #' @title Create edges parallel
 #'
-#' @description This function part of the creation of edges
+#' @description This function is part of the creation of edges
 #'
 #' @include create_edges_progress.R
 #'
-#' @param index Indices
-#' @param frags Fragments
-#' @param precs Precursors
-#' @param nspecs Number of spectra
-#'
 #' @return NULL
 #'
 #' @export
 #'
 #' @examples NULL
-create_edges_parallel <- function(index,
-                                  frags,
-                                  precs,
-                                  nspecs) {
-  list <- lapply(
-    X = (index + 1):nspecs,
-    FUN = create_edges_progress,
-    query = index,
-    s1 = cbind(mz = frags[[index]][, 1], intensity = frags[[index]][, 2]),
-    frags = frags,
-    precs = precs
-  ) |>
-    tidytable::bind_rows()
-  return(list)
+create_edges_parallel <- function(spectra,
+                                  ms2_tolerance =
+                                    params$ms$tolerances$mass$dalton$ms2,
+                                  ppm_tolerance =
+                                    params$ms$tolerances$mass$ppm$ms2,
+                                  parallel =
+                                    params$options$parallel) {
+  nspecz <- length(spectra)
+  precz <- spectra$precursorMz
+  fragz <- spectra@backend@peaksData
+  msz <- ms2_tolerance
+  ppmz <- ppm_tolerance
+
+  if (parallel) {
+    result_list <-
+      pbmcapply::pbmclapply(
+        X = 1:(nspecz - 1),
+        FUN = create_edges_progress,
+        mc.cores = parallelly::availableCores(),
+        frags = fragz,
+        precs = precz,
+        nspecs = nspecz,
+        ms2_tolerance = msz,
+        ppm_tolerance = ppmz
+      )
+  } else {
+    result_list <-
+      lapply(
+        X = 1:(nspecz - 1),
+        FUN = create_edges_progress,
+        frags = fragz,
+        precs = precz,
+        nspecs = nspecz,
+        ms2_tolerance = msz,
+        ppm_tolerance = ppmz
+      )
+  }
+
+  # Combine the results into a single matrix
+  edges <-
+    do.call(rbind, unlist(result_list, recursive = FALSE)) |>
+    data.frame()
+  return(edges)
 }

R/create_edges_progress.R

@@ -6,43 +6,50 @@ utils::globalVariables(
 
 #' @title Create edges progress
 #'
-#' @description This function is slow so it outputs
-#'    the progression of the creation of edges
+#' @description This function is slow so it had to be parallelized
 #'
-#' @param target Indices of target spectra
-#' @param query Index of query spectra
-#' @param s1 Query spectrum
+#' @param index Indices
 #' @param frags Fragments
 #' @param precs Precursors
+#' @param nspecs Number of spectra
 #'
 #' @return NULL
 #'
 #' @export
 #'
 #' @examples NULL
-create_edges_progress <- function(target,
-                                  query,
-                                  s1,
+create_edges_progress <- function(index,
                                   frags,
-                                  precs) {
-  s2 <- cbind(mz = frags[[target]][, 1], intensity = frags[[target]][, 2])
-  map <-
-    MsCoreUtils::join_gnps(
+                                  precs,
+                                  nspecs,
+                                  ms2_tolerance,
+                                  ppm_tolerance) {
+  s1 <- frags[[index]]
+  query_prec <- precs[index]
+
+  inner_list <- list()
+
+  for (target in (index + 1):nspecs) {
+    s2 <- frags[[target]]
+    map <- MsCoreUtils::join_gnps(
       x = s1[, 1],
       y = s2[, 1],
-      xPrecursorMz = precs[query],
+      xPrecursorMz = query_prec,
       yPrecursorMz = precs[target],
-      tolerance = params$ms$tolerances$mass$dalton$ms2,
-      ppm = params$ms$tolerances$mass$ppm$ms2
+      tolerance = ms2_tolerance,
+      ppm = ppm_tolerance
     )
-  matched_peaks_count <- length((map$x * map$y)[!is.na(map$x * map$y)])
-  return(
-    data.frame(
-      "feature_id" = query,
+    xy_product <- map$x * map$y
+    matched_peaks_count <- sum(!is.na(xy_product))
+
+    inner_list[[target - index]] <- list(
+      "feature_id" = index,
       "target_id" = target,
       "score" = MsCoreUtils::gnps(s1[map$x, ], s2[map$y, ]),
       "matched_peaks_count" = matched_peaks_count,
       "presence_ratio" = matched_peaks_count / length(map$y)
     )
-  )
+  }
+
+  return(inner_list)
 }

R/create_edges_spectra.R

@@ -91,29 +91,8 @@ create_edges_spectra <- function(
       expect a long processing time."
     )
     log_debug("Take yourself a break, you deserve it.")
-    nspecz <- length(spectra)
-    precz <- spectra$precursorMz
-    fragz <- spectra@backend@peaksData
-    ## Originally written with future but too slow...TODO investigate
-    ## TODO investigate weird error
-    zzz <- lapply(
-      X = 2:1,
-      FUN = create_edges_progress,
-      query = 1,
-      s1 = cbind(mz = fragz[[1]][, 1], intensity = fragz[[1]][, 2]),
-      frags = fragz,
-      precs = precz
-    )
-    edges <- pbmcapply::pbmclapply(
-      X = 1:(nspecz - 1),
-      mc.cores = parallelly::availableCores(),
-      ignore.interactive = TRUE,
-      FUN = create_edges_parallel,
-      frags = fragz,
-      precs = precz,
-      nspecs = nspecz
-    ) |>
-      tidytable::bind_rows()
+    edges <- spectra |>
+      create_edges_parallel()
 
     ## old version, keep in case
     # par <- if (parallel) {

inst/pipelines/_targets.R

@@ -2393,7 +2393,7 @@ list(
             ratio,
           condition = def_ann_spe$annotations$ms2$thresholds$condition,
           qutoff = 0,
-          parallel = def_ann_spe$options$parallel,
+          parallel = TRUE,
           fast = def_ann_spe$options$fast,
           approx = def_ann_spe$annotations$ms2$approx,
           parameters = def_ann_spe
@@ -2426,7 +2426,7 @@ list(
             ratio,
           condition = def_ann_spe$annotations$ms2$thresholds$condition,
           qutoff = 0,
-          parallel = def_ann_spe$options$parallel,
+          parallel = TRUE,
           fast = def_ann_spe$options$fast,
           approx = def_ann_spe$annotations$ms2$approx,
           parameters = def_ann_spe

tests/testthat/test_functions.R

@@ -293,28 +293,15 @@ testthat::test_that("Whole process", {
     fast = FALSE,
     condition = "AND"
   )
+  ### Semi slow
+  create_edges_spectra(
+    parallel = TRUE,
+    fast = FALSE,
+    condition = "AND"
+  )
   ### Normal
   create_edges_spectra(condition = "OR")
 
-  ## additional test not covered by lapply
-  spectra <- params$files$spectral$raw |>
-    import_spectra()
-  spectra <- spectra |>
-    sanitize_spectra(cutoff = params$ms$intensity$thresholds$ms2)
-  frags <- spectra@backend@peaksData[1:2]
-  # single_pair[[1]] <- single_pair[[1]] |>
-  #   normalize_peaks()
-  # single_pair[[2]] <- single_pair[[2]] |>
-  #   normalize_peaks()
-  precs <- spectra$precursorMz[1:2]
-  nspecs <- length(frags)
-  create_edges_parallel(1,
-    frags = frags,
-    precs = precs,
-    nspecs = nspecs
-  )
-  ##
-
   ### GNPS results
   step <- "prepare_annotations_gnps"
   params <- get_params(step = step)