option to run sequentially. various fixes. R CMD check OK

[epclust.git] / epclust / R / main.R

diff --git a/epclust/R/main.R b/epclust/R/main.R
index a982f4c..9064dfa 100644 (file)
--- a/epclust/R/main.R
+++ b/epclust/R/main.R
@@ -30,6 +30,7 @@
 #' @param nbytes Number of bytes to serialize a floating-point number; 4 or 8
 #' @param endian Endianness to use for (de)serialization. Use "little" or "big" for portability
 #' @param verbose Level of verbosity (0/FALSE for nothing or 1/TRUE for all; devel stage)
+#' @param parll TRUE to fully parallelize; otherwise run sequentially (debug, comparison)
 #'
 #' @return A matrix of the final medoids curves (K2) in rows
 #'
@@ -104,13 +105,14 @@ claws = function(getSeries, K1, K2,
        nb_series_per_chunk=50*K1, min_series_per_chunk=5*K1, #chunk size
        sep=",", #ASCII input separator
        nbytes=4, endian=.Platform$endian, #serialization (write,read)
-       verbose=FALSE)
+       verbose=FALSE, parll=TRUE)
 {
        # Check/transform arguments
-       if (!is.matrix(getSeries) && !is.function(getSeries) &&
-               !methods::is(getSeries, "connection" && !is.character(getSeries)))
+       if (!is.matrix(getSeries) && !bigmemory::is.big.matrix(getSeries)
+               && !is.function(getSeries)
+               && !methods::is(getSeries,"connection") && !is.character(getSeries))
        {
-               stop("'getSeries': matrix, function, file or valid connection (no NA)")
+               stop("'getSeries': [big]matrix, function, file or valid connection (no NA)")
        }
        K1 = .toInteger(K1, function(x) x>=2)
        K2 = .toInteger(K2, function(x) x>=2)
@@ -148,16 +150,9 @@ claws = function(getSeries, K1, K2,
        nb_curves = 0
        if (verbose)
                cat("...Compute contributions and serialize them\n")
-       repeat
-       {
-               series = getSeries((index-1)+seq_len(nb_series_per_chunk))
-               if (is.null(series))
-                       break
-               contribs_chunk = curvesToContribs(series, wf, ctype)
-               binarize(contribs_chunk, contribs_file, nb_series_per_chunk, sep, nbytes, endian)
-               index = index + nb_series_per_chunk
-               nb_curves = nb_curves + nrow(contribs_chunk)
-       }
+       nb_curves = binarizeTransform(getSeries,
+               function(series) curvesToContribs(series, wf, ctype),
+               contribs_file, nb_series_per_chunk, nbytes, endian)
        getContribs = function(indices) getDataInFile(indices, contribs_file, nbytes, endian)
 
        if (nb_curves < min_series_per_chunk)
@@ -174,28 +169,37 @@ claws = function(getSeries, K1, K2,
        })
        if (verbose)
                cat(paste("...Run ",ntasks," x stage 1 in parallel\n",sep=""))
-#      cl = parallel::makeCluster(ncores_tasks)
-#      parallel::clusterExport(cl, varlist=c("getSeries","getContribs","K1","K2",
-#              "nb_series_per_chunk","ncores_clust","synchrones_file","sep","nbytes","endian"),
-#              envir = environment())
-       # 1000*K1 indices [if WER=="end"], or empty vector [if WER=="mix"] --> series on file
-#      indices = unlist( parallel::parLapply(cl, indices_tasks, function(inds) {
-       indices = unlist( lapply(indices_tasks, function(inds) {
-#              require("epclust", quietly=TRUE)
-
-               browser() #TODO: CONTINUE DEBUG HERE
+       if (parll)
+       {
+               cl = parallel::makeCluster(ncores_tasks)
+               parallel::clusterExport(cl, varlist=c("getSeries","getContribs","K1","K2","verbose","parll",
+                       "nb_series_per_chunk","ncores_clust","synchrones_file","sep","nbytes","endian"),
+                       envir = environment())
+       }
 
-               indices_medoids = clusteringTask(inds,getContribs,K1,nb_series_per_chunk,ncores_clust)
+       runTwoStepClustering = function(inds)
+       {
+               if (parll)
+                       require("epclust", quietly=TRUE)
+               indices_medoids = clusteringTask1(
+                       inds, getContribs, K1, nb_series_per_chunk, ncores_clust, verbose, parll)
                if (WER=="mix")
                {
-                       medoids2 = computeClusters2(
-                               getSeries(indices_medoids), K2, getSeries, nb_series_per_chunk)
+                       medoids2 = computeClusters2(getSeries(indices_medoids),
+                               K2, getSeries, nb_curves, nb_series_per_chunk, ncores_clust, verbose, parll)
                        binarize(medoids2, synchrones_file, nb_series_per_chunk, sep, nbytes, endian)
                        return (vector("integer",0))
                }
                indices_medoids
-       }) )
-#      parallel::stopCluster(cl)
+       }
+
+       # 1000*K1 indices [if WER=="end"], or empty vector [if WER=="mix"] --> series on file
+       if (parll)
+               indices = unlist( parallel::parLapply(cl, indices_tasks, runTwoStepClustering) )
+       else
+               indices = unlist( lapply(indices_tasks, runTwoStepClustering) )
+       if (parll)
+               parallel::stopCluster(cl)
 
        getRefSeries = getSeries
        synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file)
@@ -209,23 +213,18 @@ claws = function(getSeries, K1, K2,
                index = 1
                if (verbose)
                        cat("...Serialize contributions computed on synchrones\n")
-               repeat
-               {
-                       series = getSeries((index-1)+seq_len(nb_series_per_chunk))
-                       if (is.null(series))
-                               break
-                       contribs_chunk = curvesToContribs(series, wf, ctype)
-                       binarize(contribs_chunk, contribs_file, nb_series_per_chunk, sep, nbytes, endian)
-                       index = index + nb_series_per_chunk
-               }
+               ignored = binarizeTransform(getSeries,
+                       function(series) curvesToContribs(series, wf, ctype),
+                       contribs_file, nb_series_per_chunk, nbytes, endian)
        }
 
        # Run step2 on resulting indices or series (from file)
        if (verbose)
                cat("...Run final // stage 1 + stage 2\n")
-       indices_medoids = clusteringTask(
-               indices, getContribs, K1, nb_series_per_chunk, ncores_tasks*ncores_clust)
-       medoids = computeClusters2(getSeries(indices_medoids),K2,getRefSeries,nb_series_per_chunk)
+       indices_medoids = clusteringTask1(
+               indices, getContribs, K1, nb_series_per_chunk, ncores_tasks*ncores_clust, verbose)
+       medoids = computeClusters2(getSeries(indices_medoids),
+               K2, getRefSeries, nb_curves, nb_series_per_chunk, ncores_tasks*ncores_clust, verbose)
 
        # Cleanup
        unlink(bin_dir, recursive=TRUE)
@@ -259,7 +258,7 @@ curvesToContribs = function(series, wf, ctype)
        }) )
 }
 
-# Helper for main function: check integer arguments with functiional conditions
+# Check integer arguments with functional conditions
 .toInteger <- function(x, condition)
 {
        if (!is.integer(x))