X-Git-Url: https://git.auder.net/?a=blobdiff_plain;f=epclust%2FR%2Fmain.R;h=977e61b235da2ea1ac4e0fc3c4cd40bbddcc049f;hb=e161499b97c782aadfc287c22b55f85724f86fae;hp=b09e93418836dd5786037323dca1d94538a2e5b4;hpb=4bcfdbee4e2157f232427a5bfdf240f34760110d;p=epclust.git diff --git a/epclust/R/main.R b/epclust/R/main.R index b09e934..977e61b 100644 --- a/epclust/R/main.R +++ b/epclust/R/main.R @@ -7,8 +7,9 @@ #' @param getSeries Access to the (time-)series, which can be of one of the three #' following types: #' \itemize{ -#' \item matrix: each line contains all the values for one time-serie, ordered by time -#' \item connection: any R connection object (e.g. a file) providing lines as described above +#' \item [big.]matrix: each line contains all the values for one time-serie, ordered by time +#' \item connection: any R connection object providing lines as described above +#' \item character: name of a CSV file containing series in rows (no header) #' \item function: a custom way to retrieve the curves; it has only one argument: #' the indices of the series to be retrieved. See examples #' } @@ -30,8 +31,9 @@ #' @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 +#' @return A big.matrix of the final medoids curves (K2) in rows #' #' @examples #' \dontrun{ @@ -104,13 +106,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) @@ -131,7 +134,7 @@ claws = function(getSeries, K1, K2, nbytes = .toInteger(nbytes, function(x) x==4 || x==8) # Serialize series if required, to always use a function - bin_dir = ".epclust.bin/" + bin_dir = ".epclust_bin/" dir.create(bin_dir, showWarnings=FALSE, mode="0755") if (!is.function(getSeries)) { @@ -142,22 +145,15 @@ claws = function(getSeries, K1, K2, getSeries = function(inds) getDataInFile(inds, series_file, nbytes, endian) } - # Serialize all computed wavelets contributions onto a file + # Serialize all computed wavelets contributions into a file contribs_file = paste(bin_dir,"contribs",sep="") ; unlink(contribs_file) index = 1 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) @@ -166,6 +162,23 @@ claws = function(getSeries, K1, K2, if (nb_series_per_task < min_series_per_chunk) stop("Too many tasks: less series in one task than min_series_per_chunk!") + runTwoStepClustering = function(inds) + { + if (parll && ntasks>1) + require("epclust", quietly=TRUE) + indices_medoids = clusteringTask1( + inds, getContribs, K1, nb_series_per_chunk, ncores_clust, verbose, parll) + if (WER=="mix") + { + medoids1 = bigmemory::as.big.matrix( getSeries(indices_medoids) ) + medoids2 = clusteringTask2(medoids1, + 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 + } + # Cluster contributions in parallel (by nb_series_per_chunk) indices_all = if (random) sample(nb_curves) else seq_len(nb_curves) indices_tasks = lapply(seq_len(ntasks), function(i) { @@ -173,32 +186,33 @@ claws = function(getSeries, K1, K2, indices_all[((i-1)*nb_series_per_task+1):upper_bound] }) 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 - - indices_medoids = clusteringTask(inds,getContribs,K1,nb_series_per_chunk,ncores_clust) + { + message = paste("...Run ",ntasks," x stage 1", sep="") if (WER=="mix") - { - medoids2 = computeClusters2( - getSeries(indices_medoids), K2, getSeries, nb_series_per_chunk) - binarize(medoids2, synchrones_file, nb_series_per_chunk, sep, nbytes, endian) - return (vector("integer",0)) - } - indices_medoids - }) ) -# parallel::stopCluster(cl) + message = paste(message," + stage 2", sep="") + cat(paste(message,"\n", sep="")) + } + if (WER=="mix") + {synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file)} + if (parll && ntasks>1) + { + cl = parallel::makeCluster(ncores_tasks) + varlist = c("getSeries","getContribs","K1","K2","verbose","parll", + "nb_series_per_chunk","ntasks","ncores_clust","sep","nbytes","endian") + if (WER=="mix") + varlist = c(varlist, "synchrones_file") + parallel::clusterExport(cl, varlist=varlist, envir = environment()) + } + + # 1000*K1 indices [if WER=="end"], or empty vector [if WER=="mix"] --> series on file + if (parll && ntasks>1) + indices = unlist( parallel::parLapply(cl, indices_tasks, runTwoStepClustering) ) + else + indices = unlist( lapply(indices_tasks, runTwoStepClustering) ) + if (parll && ntasks>1) + parallel::stopCluster(cl) getRefSeries = getSeries - synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file) if (WER=="mix") { indices = seq_len(ntasks*K2) @@ -209,28 +223,24 @@ 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, parll) + medoids1 = bigmemory::as.big.matrix( getSeries(indices_medoids) ) + medoids2 = clusteringTask2(medoids1, K2, + getRefSeries, nb_curves, nb_series_per_chunk, ncores_tasks*ncores_clust, verbose, parll) # Cleanup unlink(bin_dir, recursive=TRUE) - medoids + medoids2 } #' curvesToContribs @@ -259,7 +269,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))