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 (nb_series_per_task < min_series_per_chunk)
stop("Too many tasks: less series in one task than min_series_per_chunk!")
- # 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) {
- upper_bound = ifelse( i<ntasks, min(nb_series_per_task*i,nb_curves), nb_curves )
- indices_all[((i-1)*nb_series_per_task+1):upper_bound]
- })
- if (verbose)
- cat(paste("...Run ",ntasks," x stage 1 in parallel\n",sep=""))
- 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())
- }
-
runTwoStepClustering = function(inds)
{
if (parll)
inds, getContribs, K1, nb_series_per_chunk, ncores_clust, verbose, parll)
if (WER=="mix")
{
+
+
+
+
+#TODO: getSeries(indices_medoids) BAD ; il faudrait une big.matrix de medoids en entree
+ #OK en RAM il y en aura 1000 (donc 1000*K1*17519... OK)
+ #...mais du coup chaque process ne re-dupliquera pas medoids
+
+
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)
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) {
+ upper_bound = ifelse( i<ntasks, min(nb_series_per_task*i,nb_curves), nb_curves )
+ indices_all[((i-1)*nb_series_per_task+1):upper_bound]
+ })
+ if (verbose)
+ cat(paste("...Run ",ntasks," x stage 1 in parallel\n",sep=""))
+ if (WER=="mix")
+ {synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file)}
+ if (parll)
+ {
+ cl = parallel::makeCluster(ncores_tasks)
+ varlist = c("getSeries","getContribs","K1","K2","verbose","parll",
+ "nb_series_per_chunk","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)
indices = unlist( parallel::parLapply(cl, indices_tasks, runTwoStepClustering) )
parallel::stopCluster(cl)
getRefSeries = getSeries
- synchrones_file = paste(bin_dir,"synchrones",sep="") ; unlink(synchrones_file)
if (WER=="mix")
{
indices = seq_len(ntasks*K2)
contribs_file, nb_series_per_chunk, nbytes, endian)
}
+
+
+#TODO: if ntasks==1, c'est deja terminé
+
# Run step2 on resulting indices or series (from file)
if (verbose)
cat("...Run final // stage 1 + stage 2\n")