use same package versions on cluster machines as source

R/DEoptim_fns.R

@@ -97,6 +97,7 @@
                        .verbose,
                        visualizeDEoptim,
                        .plotSize = list(height = 1600, width = 2000)) {
+  if (isTRUE(is.na(cores))) cores <- NULL
   origBlas <- blas_get_num_procs()
   if (origBlas > 1) {
     blas_set_num_threads(1)
@@ -150,61 +151,79 @@
     ## Make cluster with just one worker per machine --> don't need to do these steps
     #     multiple times per machine, if not all 'localhost'
     revtunnel <- FALSE
+    neededPkgs <- c("kSamples", "magrittr", "raster", "data.table",
+                    "SpaDES.tools", "fireSenseUtils")
+
+    # browser()
     if (!identical("localhost", unique(cores))) {
-      revtunnel <- ifelse(all(cores == "localhost"), FALSE, TRUE)
+      repos <- c("https://predictiveecology.r-universe.dev", getOption("repos"))
 
-      coresUnique <- setdiff(unique(cores), "localhost")
-      message(
-        "Making sure packages with sufficient versions installed and loaded on: ",
-        paste(coresUnique, collapse = ", ")
-      )
-      st <- system.time({
-        cl <- parallelly::makeClusterPSOCK(coresUnique, revtunnel = revtunnel, rscript_libs = libPath)
-      })
-      packageVersionFSU <- packageVersion("fireSenseUtils")
-      packageVersionST <- packageVersion("SpaDES.tools")
-      clusterExport(cl, list("libPath", "logPath", "packageVersionFSU", "packageVersionST"), envir = environment())
-
-      parallel::clusterEvalQ(
-        cl,
-        {
-          # If this is first time that packages need to be installed for this user on this machine
-          #   there won't be a folder present that is writable
-          if (!dir.exists(libPath)) {
-            dir.create(libPath, recursive = TRUE)
+      aa <- Require::pkgDep(unique(c("dqrng", "PredictiveEcology/SpaDES.tools@development",
+                                     "PredictiveEcology/fireSenseUtils@development", "qs", "RCurl", neededPkgs)), recursive = TRUE)
+      revtunnel <- ifelse(all(cores == "localhost"), FALSE, TRUE)
 
+        coresUnique <- setdiff(unique(cores), "localhost")
+        message(
+          "Making sure packages with sufficient versions installed and loaded on: ",
+          paste(coresUnique, collapse = ", ")
+        )
+        st <- system.time({
+          cl <- parallelly::makeClusterPSOCK(coresUnique, revtunnel = revtunnel, rscript_libs = libPath)
+        })
+        packageVersionFSU <- packageVersion("fireSenseUtils")
+        packageVersionST <- packageVersion("SpaDES.tools")
+        clusterExport(cl, list("libPath", "logPath", "packageVersionFSU", "packageVersionST", "repos"),
+                      envir = environment())
+
+        parallel::clusterEvalQ(
+          cl,
+          {
+            # If this is first time that packages need to be installed for this user on this machine
+            #   there won't be a folder present that is writable
             if (!dir.exists(libPath)) {
-              stop("libPath directory creation failed.\n",
-                   "Try creating on each machine manually, using e.g.,\n",
-                   "  mkdir -p ", libPath)
+              dir.create(libPath, recursive = TRUE)
+
+              if (!dir.exists(libPath)) {
+                stop("libPath directory creation failed.\n",
+                     "Try creating on each machine manually, using e.g.,\n",
+                     "  mkdir -p ", libPath)
+              }
             }
-          }
 
-          if (!"Require" %in% rownames(utils::installed.packages())) {
-            remotes::install_github("PredictiveEcology/Require@development")
-          } else if (packageVersion("Require") < "0.1.0.9000") {
-            remotes::install_github("PredictiveEcology/Require@development")
-          }
+            # logPath <- Require::checkPath(dirname(logPath), create = TRUE)
+            logPath <- dir.create(dirname(logPath), showWarnings = FALSE, recursive = TRUE)
+
+            message(Sys.info()[["nodename"]])
 
-          ## Use the binary packages for install if Ubuntu & Linux
-          Require::setLinuxBinaryRepo()
+            #scp -r /home/emcintir/.local/share/R/Edehzhie/packages/x86_64-pc-linux-gnu/4.3/fireSenseUtils emcintir@10.│^C
+            #20.0.97:/home/emcintir/.local/share/R/Edehzhie/packages/x86_64-pc-linux-gnu/4.3
 
-          logPath <- Require::checkPath(dirname(logPath), create = TRUE)
+            needRequire <- FALSE
+            if (!"Require" %in% rownames(utils::installed.packages())) {
+              needRequire <- TRUE
+            } else if (packageVersion("Require") < "0.1.0.9000")  {
+              needRequire <- TRUE
+            }
+            if (isTRUE(needRequire))
+              install.packages("Require", repos = repos, lib = libPath)
 
-          message(Sys.info()[["nodename"]])
+            ## Use the binary packages for install if Ubuntu & Linux
+            # Require::setLinuxBinaryRepo()
 
-          ## This will install the versions of SpaDES.tools and fireSenseUtils that are on the main machine
-          Require::Require(
-            c(
-              "dqrng",
-              paste0("PredictiveEcology/SpaDES.tools@development (>=", packageVersionST, ")"),
-              paste0("PredictiveEcology/fireSenseUtils@development (>=", packageVersionFSU, ")")
-            ),
-            upgrade = FALSE
-          )
-        }
-      )
-      parallel::stopCluster(cl)
+            if (FALSE) {
+              ## This will install the versions of SpaDES.tools and fireSenseUtils that are on the main machine
+              Require::Require(c("dqrng", "SpaDES.tools", "fireSenseUtils"), repos = repos)
+            }
+
+          }
+        )
+        pkgsNeeded <- unique(Require::extractPkgName(unname(unlist(aa))))
+        out <- lapply(setdiff(unique(cores), "localhost"), function(ip) {
+          system(paste0("rsync -aruv --update ", paste(file.path(libPath, pkgsNeeded), collapse = " "),
+                        " ", ip, ":", libPath))
+        })
+
+        parallel::stopCluster(cl)
     }
 
     ## Now make full cluster with one worker per core listed in "cores"
@@ -237,14 +256,14 @@
           }
           x
         })
-        filenameForTransfer <- Require::normPath(tempfile(fileext = ".qs"))
-        Require::checkPath(dirname(filenameForTransfer), create = TRUE) # during development, this was deleted accidentally
+        filenameForTransfer <- normalizePath(tempfile(fileext = ".qs"), mustWork = FALSE, winslash = "/")
+        dir.create(dirname(filenameForTransfer), recursive = TRUE, showWarnings = FALSE) # during development, this was deleted accidentally
         qs::qsave(objsToCopy, file = filenameForTransfer)
         stExport <- system.time({
           outExp <- clusterExport(cl, varlist = "filenameForTransfer", envir = environment())
         })
         out11 <- clusterEvalQ(cl, {
-          Require::checkPath(dirname(filenameForTransfer), create = TRUE)
+          dir.create(dirname(filenameForTransfer), recursive = TRUE, showWarnings = FALSE)
         })
         out <- lapply(setdiff(unique(cores), "localhost"), function(ip) {
           st1 <- system.time(system(paste0("rsync -a ", filenameForTransfer, " ", ip, ":", filenameForTransfer)))
@@ -277,13 +296,12 @@
     }
     message("it took ", round(stMoveObjects[3], 2), "s to move objects to nodes")
     message("loading packages in cluster nodes")
+
+    clusterExport(cl, "neededPkgs", envir = environment())
     stPackages <- system.time(parallel::clusterEvalQ(
       cl,
       {
-        for (i in c(
-          "kSamples", "magrittr", "raster", "data.table",
-          "SpaDES.tools", "fireSenseUtils"
-        )) {
+        for (i in neededPkgs) {
           library(i, character.only = TRUE)
         }
         message("loading ", i, " at ", Sys.time())
@@ -417,6 +435,7 @@
           mutuallyExclusive = mutuallyExclusive,
           doAssertions = doObjFunAssertions,
           Nreps = Nreps,
+          plot.it = FALSE,
           controlForCache = controlForCache,
           objFunCoresInternal = objFunCoresInternal,
           thresh = thresh,

R/cleanUpSpreadFirePoints.R

@@ -19,6 +19,8 @@
 #' @importFrom sf st_as_sf st_crs
 #' @importFrom terra extract xyFromCell
 cleanUpSpreadFirePoints <- function(firePoints, bufferDT, flammableRTM) {
+  if (is.data.table(bufferDT))
+    bufferDT <- as.data.table(bufferDT)
   FlamPoints <- as.data.table(extract(flammableRTM, firePoints, cells = TRUE))
   setnames(FlamPoints, c("ID", "isFlammable", "cells"))
   FlamPoints[, isFlammable := as.numeric(as.character(isFlammable))] #otherwise factor = 1 and 2

R/objFunSpread.R

@@ -402,7 +402,7 @@
   medSPRight <- medSP <= maxFireSpread & medSP >= lowerSpreadProb
   spreadOutEnough <- sdSP / medSP > 0.025
   ret <- list()
-  minLik <- 1e-19 # min(emp$lik[emp$lik > 0])
+  minLik <- 1e-29 # min(emp$lik[emp$lik > 0])
   loci <- annualFires$cells
   summ <- summary(nonEdgeValues)
   lowSPLowEnough <- summ[2] < lanscape1stQuantileThresh
@@ -424,8 +424,6 @@
     }
   }
 
-  # att <- try(if (medSPRight && spreadOutEnough) { "hi" })
-  # if (is(att, "try-error")) browser()
   if (medSPRight && spreadOutEnough && lowSPLowEnough) {
     if (verbose) {
       ww <- if (isTRUE(weighted)) "weighted" else "unweighted"
@@ -443,6 +441,8 @@
     minSize <- 100
     if (doAssertions || plot.it) {
       tableOfBufferedMaps <- annualFireBufferedDT[, list(numAvailPixels = .N), by = "ids"]
+      tableOfBufferedMaps <- tableOfBufferedMaps[annualFires, on = "ids"]
+      setnames(tableOfBufferedMaps, old = "cells", new = "initialPixels")
       minSizes <- tableOfBufferedMaps$numAvailPixels
       minSize <- quantile(minSizes, 0.3)
       if (minSize < 2000) {
@@ -463,17 +463,20 @@
       maxSizes <- maxSizes[!dups]
       loci <- annualFires$cells[!dups]
     }
-    spreadState <- lapply(seq_len(Nreps), function(i) {
-      SpaDES.tools::spread2(
+    st <- system.time(spreadState <- lapply(seq_len(Nreps), function(i) {
+      SpaDES.tools::spread(
+        # SpaDES.tools::spread2(
         landscape = r,
         maxSize = maxSizes,
-        start = loci,
+        # start = loci,
+        loci = loci,
         spreadProb = cells,
-        asRaster = FALSE,
+        # asRaster = FALSE,
+        returnIndices = TRUE,
         allowOverlap = FALSE,
         skipChecks = TRUE
       )
-    })
+    }))
     if (isTRUE(plot.it)) {
       par(
         mfrow = c(7, 7), omi = c(0.5, 0, 0, 0),
@@ -491,11 +494,16 @@
       )
     }
     spreadState <- rbindlist(spreadState, idcol = "rep")
+    if ("indices" %in% colnames(spreadState))
+      setnames(spreadState, old = "indices", "pixels")
+    if ("initialLocus" %in% colnames(spreadState) )
+      setnames(spreadState, old = "initialLocus", "initialPixels")
     if (isTRUE(doSNLL_FSTest)) {
       emp <- spreadState[, list(N = .N), by = c("rep", "initialPixels")]
       emp <- emp[annualFires, on = c("initialPixels" = "cells")]
       if (plot.it) {
-        emp <- tableOfBufferedMaps[emp, on = c("initialPixels"), nomatch = NULL]
+        colsToKeep <- c(setdiff(colnames(tableOfBufferedMaps), colnames(emp)), "initialPixels")
+        emp <- tableOfBufferedMaps[, ..colsToKeep][emp, on = c("initialPixels"), nomatch = NULL]
         maxX <- log(max(c(annualFires$size, emp$N, emp$numAvailPixels)))
         emp <- setorderv(emp, c("size"), order = -1L)
         numLargest <- 4
@@ -511,7 +519,6 @@
         } else {
           uniqueEmpIds
         }
-        if (is(sam, "try-error")) browser()
         emp[ids %in% sam,
           {
             dat <- round(log(N))
@@ -627,7 +634,7 @@
         predLiklihood <- rast(r)
         predLiklihood[out$pixelID] <- predictedLiklihood
         predLiklihood <- crop(predLiklihood, ex)
-        spIgnits <- SpatialPoints(coords = xyFromCell(r, thisFire$cells))
+        spIgnits <- terra::vect(xyFromCell(r, thisFire$cells))
         spIgnits <- buffer(spIgnits, width = 5000)
         spIgnits <- crop(spIgnits, ex)
         list(