X-Git-Url: https://git.auder.net/?p=epclust.git;a=blobdiff_plain;f=epclust%2FR%2Fmain.R;h=9064dfaec6ee0bed2fa0a5544a093bab91141e5e;hp=a982f4cae4b671f15217e8c04b971f7a72363df3;hb=492cd9e74a79cbcc0ecde55fa3071a44b7e463dc;hpb=dc646a9749509ce4c52079d456f6967e9ac833b5

diff --git a/epclust/R/main.R b/epclust/R/main.R
index a982f4c..9064dfa 100644
--- 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))