Commit | Line | Data |
---|---|---|
4bcfdbee BA |
1 | #' @name clustering |
2 | #' @rdname clustering | |
492cd9e7 | 3 | #' @aliases clusteringTask1 computeClusters1 computeClusters2 |
4bcfdbee | 4 | #' |
492cd9e7 | 5 | #' @title Two-stage clustering, withing one task (see \code{claws()}) |
4bcfdbee | 6 | #' |
492cd9e7 BA |
7 | #' @description \code{clusteringTask1()} runs one full stage-1 task, which consists in |
8 | #' iterated stage 1 clustering (on nb_curves / ntasks energy contributions, computed | |
bf5c0844 BA |
9 | #' through discrete wavelets coefficients). |
10 | #' \code{clusteringTask2()} runs a full stage-2 task, which consists in synchrones | |
11 | #' and then WER distances computations, before applying the clustering algorithm. | |
12 | #' \code{computeClusters1()} and \code{computeClusters2()} correspond to the atomic | |
13 | #' clustering procedures respectively for stage 1 and 2. The former applies the | |
14 | #' clustering algorithm (PAM) on a contributions matrix, while the latter clusters | |
15 | #' a chunk of series inside one task (~max nb_series_per_chunk) | |
4bcfdbee BA |
16 | #' |
17 | #' @param indices Range of series indices to cluster in parallel (initial data) | |
18 | #' @param getContribs Function to retrieve contributions from initial series indices: | |
19 | #' \code{getContribs(indices)} outpus a contributions matrix | |
20 | #' @param contribs matrix of contributions (e.g. output of \code{curvesToContribs()}) | |
21 | #' @inheritParams computeSynchrones | |
22 | #' @inheritParams claws | |
23 | #' | |
492cd9e7 | 24 | #' @return For \code{clusteringTask1()} and \code{computeClusters1()}, the indices of the |
4bcfdbee | 25 | #' computed (K1) medoids. Indices are irrelevant for stage 2 clustering, thus |
24ed5d83 | 26 | #' \code{computeClusters2()} outputs a big.matrix of medoids |
4bcfdbee BA |
27 | #' (of size limited by nb_series_per_chunk) |
28 | NULL | |
29 | ||
30 | #' @rdname clustering | |
31 | #' @export | |
492cd9e7 BA |
32 | clusteringTask1 = function( |
33 | indices, getContribs, K1, nb_series_per_chunk, ncores_clust=1, verbose=FALSE, parll=TRUE) | |
5c652979 | 34 | { |
492cd9e7 BA |
35 | if (verbose) |
36 | cat(paste("*** Clustering task on ",length(indices)," lines\n", sep="")) | |
4bcfdbee | 37 | |
492cd9e7 BA |
38 | wrapComputeClusters1 = function(inds) { |
39 | if (parll) | |
40 | require("epclust", quietly=TRUE) | |
41 | if (verbose) | |
42 | cat(paste(" computeClusters1() on ",length(inds)," lines\n", sep="")) | |
43 | inds[ computeClusters1(getContribs(inds), K1) ] | |
44 | } | |
4bcfdbee | 45 | |
492cd9e7 | 46 | if (parll) |
7b13d0c2 | 47 | { |
492cd9e7 BA |
48 | cl = parallel::makeCluster(ncores_clust) |
49 | parallel::clusterExport(cl, varlist=c("getContribs","K1","verbose"), envir=environment()) | |
7b13d0c2 | 50 | } |
492cd9e7 BA |
51 | while (length(indices) > K1) |
52 | { | |
53 | indices_workers = .spreadIndices(indices, nb_series_per_chunk) | |
54 | if (parll) | |
55 | indices = unlist( parallel::parLapply(cl, indices_workers, wrapComputeClusters1) ) | |
56 | else | |
57 | indices = unlist( lapply(indices_workers, wrapComputeClusters1) ) | |
58 | } | |
59 | if (parll) | |
60 | parallel::stopCluster(cl) | |
61 | ||
56857861 | 62 | indices #medoids |
5c652979 BA |
63 | } |
64 | ||
4bcfdbee BA |
65 | #' @rdname clustering |
66 | #' @export | |
bf5c0844 | 67 | clusteringTask2 = function(medoids, K2, |
492cd9e7 | 68 | getRefSeries, nb_ref_curves, nb_series_per_chunk, ncores_clust=1,verbose=FALSE,parll=TRUE) |
5c652979 | 69 | { |
bf5c0844 BA |
70 | if (nrow(medoids) <= K2) |
71 | return (medoids) | |
492cd9e7 BA |
72 | synchrones = computeSynchrones(medoids, |
73 | getRefSeries, nb_ref_curves, nb_series_per_chunk, ncores_clust, verbose, parll) | |
74 | distances = computeWerDists(synchrones, ncores_clust, verbose, parll) | |
bf5c0844 BA |
75 | # PAM in package 'cluster' cannot take big.matrix in input: need to cast it |
76 | mat_dists = matrix(nrow=K1, ncol=K1) | |
77 | for (i in seq_len(K1)) | |
78 | mat_dists[i,] = distances[i,] | |
79 | medoids[ computeClusters2(mat_dists,K2), ] | |
5c652979 BA |
80 | } |
81 | ||
bf5c0844 BA |
82 | #' @rdname clustering |
83 | #' @export | |
84 | computeClusters1 = function(contribs, K1) | |
85 | cluster::pam(contribs, K1, diss=FALSE)$id.med | |
86 | ||
87 | #' @rdname clustering | |
88 | #' @export | |
89 | computeClusters2 = function(distances, K2) | |
90 | cluster::pam(distances, K2, diss=TRUE)$id.med | |
91 | ||
4bcfdbee BA |
92 | #' computeSynchrones |
93 | #' | |
94 | #' Compute the synchrones curves (sum of clusters elements) from a matrix of medoids, | |
95 | #' using L2 distances. | |
96 | #' | |
24ed5d83 | 97 | #' @param medoids big.matrix of medoids (curves of same length as initial series) |
4bcfdbee BA |
98 | #' @param getRefSeries Function to retrieve initial series (e.g. in stage 2 after series |
99 | #' have been replaced by stage-1 medoids) | |
492cd9e7 | 100 | #' @param nb_ref_curves How many reference series? (This number is known at this stage) |
4bcfdbee BA |
101 | #' @inheritParams claws |
102 | #' | |
24ed5d83 BA |
103 | #' @return A big.matrix of size K1 x L where L = data_length |
104 | #' | |
4bcfdbee | 105 | #' @export |
492cd9e7 BA |
106 | computeSynchrones = function(medoids, getRefSeries, |
107 | nb_ref_curves, nb_series_per_chunk, ncores_clust=1,verbose=FALSE,parll=TRUE) | |
e205f218 | 108 | { |
492cd9e7 | 109 | computeSynchronesChunk = function(indices) |
3eef8d3d | 110 | { |
c45fd663 BA |
111 | if (verbose) |
112 | cat(paste("--- Compute synchrones for ",length(indices)," lines\n", sep="")) | |
492cd9e7 | 113 | ref_series = getRefSeries(indices) |
3eef8d3d | 114 | #get medoids indices for this chunk of series |
56857861 BA |
115 | for (i in seq_len(nrow(ref_series))) |
116 | { | |
8702eb86 | 117 | j = which.min( rowSums( sweep(medoids, 2, ref_series[i,], '-')^2 ) ) |
492cd9e7 BA |
118 | if (parll) |
119 | synchronicity::lock(m) | |
8702eb86 | 120 | synchrones[j,] = synchrones[j,] + ref_series[i,] |
492cd9e7 BA |
121 | counts[j,1] = counts[j,1] + 1 |
122 | if (parll) | |
123 | synchronicity::unlock(m) | |
124 | } | |
125 | } | |
126 | ||
127 | K = nrow(medoids) | |
128 | # Use bigmemory (shared==TRUE by default) + synchronicity to fill synchrones in // | |
24ed5d83 | 129 | # TODO: if size > RAM (not our case), use file-backed big.matrix |
492cd9e7 BA |
130 | synchrones = bigmemory::big.matrix(nrow=K,ncol=ncol(medoids),type="double",init=0.) |
131 | counts = bigmemory::big.matrix(nrow=K,ncol=1,type="double",init=0) | |
24ed5d83 | 132 | # synchronicity is only for Linux & MacOS; on Windows: run sequentially |
492cd9e7 BA |
133 | parll = (requireNamespace("synchronicity",quietly=TRUE) |
134 | && parll && Sys.info()['sysname'] != "Windows") | |
135 | if (parll) | |
136 | m <- synchronicity::boost.mutex() | |
137 | ||
24ed5d83 BA |
138 | if (parll) |
139 | { | |
140 | cl = parallel::makeCluster(ncores_clust) | |
141 | parallel::clusterExport(cl, | |
142 | varlist=c("synchrones","counts","verbose","medoids","getRefSeries"), | |
143 | envir=environment()) | |
144 | } | |
145 | ||
492cd9e7 | 146 | indices_workers = .spreadIndices(seq_len(nb_ref_curves), nb_series_per_chunk) |
c45fd663 | 147 | ignored <- |
492cd9e7 | 148 | if (parll) |
24ed5d83 | 149 | parallel::parLapply(indices_workers, computeSynchronesChunk) |
492cd9e7 | 150 | else |
c45fd663 | 151 | lapply(indices_workers, computeSynchronesChunk) |
492cd9e7 | 152 | |
24ed5d83 BA |
153 | if (parll) |
154 | parallel::stopCluster(cl) | |
155 | ||
156 | #TODO: can we avoid this loop? ( synchrones = sweep(synchrones, 1, counts, '/') ) | |
492cd9e7 | 157 | for (i in seq_len(K)) |
24ed5d83 | 158 | synchrones[i,] = synchrones[i,] / counts[i,1] |
3eef8d3d | 159 | #NOTE: odds for some clusters to be empty? (when series already come from stage 2) |
8702eb86 | 160 | # ...maybe; but let's hope resulting K1' be still quite bigger than K2 |
24ed5d83 BA |
161 | noNA_rows = sapply(seq_len(K), function(i) all(!is.nan(synchrones[i,]))) |
162 | if (all(noNA_rows)) | |
163 | return (synchrones) | |
164 | # Else: some clusters are empty, need to slice synchrones | |
165 | synchrones[noNA_rows,] | |
e205f218 | 166 | } |
1c6f223e | 167 | |
4bcfdbee BA |
168 | #' computeWerDists |
169 | #' | |
170 | #' Compute the WER distances between the synchrones curves (in rows), which are | |
171 | #' returned (e.g.) by \code{computeSynchrones()} | |
172 | #' | |
24ed5d83 BA |
173 | #' @param synchrones A big.matrix of synchrones, in rows. The series have same length |
174 | #' as the series in the initial dataset | |
492cd9e7 | 175 | #' @inheritParams claws |
4bcfdbee | 176 | #' |
24ed5d83 BA |
177 | #' @return A big.matrix of size K1 x K1 |
178 | #' | |
4bcfdbee | 179 | #' @export |
492cd9e7 | 180 | computeWerDists = function(synchrones, ncores_clust=1,verbose=FALSE,parll=TRUE) |
d03c0621 | 181 | { |
24ed5d83 BA |
182 | |
183 | ||
184 | ||
185 | #TODO: re-organize to call computeWerDist(x,y) [C] (in //?) from two indices + big.matrix | |
186 | ||
187 | ||
4bcfdbee BA |
188 | n <- nrow(synchrones) |
189 | delta <- ncol(synchrones) | |
db6fc17d | 190 | #TODO: automatic tune of all these parameters ? (for other users) |
d03c0621 | 191 | nvoice <- 4 |
4bcfdbee | 192 | # noctave = 2^13 = 8192 half hours ~ 180 days ; ~log2(ncol(synchrones)) |
d7d55bc1 BA |
193 | noctave = 13 |
194 | # 4 here represent 2^5 = 32 half-hours ~ 1 day | |
db6fc17d | 195 | #NOTE: default scalevector == 2^(0:(noctave * nvoice) / nvoice) * s0 (?) |
24ed5d83 | 196 | scalevector <- 2^(4:(noctave * nvoice) / nvoice + 1) |
db6fc17d BA |
197 | #condition: ( log2(s0*w0/(2*pi)) - 1 ) * nvoice + 1.5 >= 1 |
198 | s0=2 | |
199 | w0=2*pi | |
200 | scaled=FALSE | |
201 | s0log = as.integer( (log2( s0*w0/(2*pi) ) - 1) * nvoice + 1.5 ) | |
202 | totnoct = noctave + as.integer(s0log/nvoice) + 1 | |
203 | ||
492cd9e7 BA |
204 | computeCWT = function(i) |
205 | { | |
206 | if (verbose) | |
207 | cat(paste("+++ Compute Rwave::cwt() on serie ",i,"\n", sep="")) | |
4bcfdbee | 208 | ts <- scale(ts(synchrones[i,]), center=TRUE, scale=scaled) |
24ed5d83 | 209 | totts.cwt = Rwave::cwt(ts, totnoct, nvoice, w0, plot=FALSE) |
db6fc17d BA |
210 | ts.cwt = totts.cwt[,s0log:(s0log+noctave*nvoice)] |
211 | #Normalization | |
212 | sqs <- sqrt(2^(0:(noctave*nvoice)/nvoice)*s0) | |
af3ea947 | 213 | sqres <- sweep(ts.cwt,2,sqs,'*') |
db6fc17d | 214 | sqres / max(Mod(sqres)) |
492cd9e7 | 215 | } |
3ccd1e39 | 216 | |
492cd9e7 BA |
217 | if (parll) |
218 | { | |
219 | cl = parallel::makeCluster(ncores_clust) | |
af3ea947 BA |
220 | parallel::clusterExport(cl, |
221 | varlist=c("synchrones","totnoct","nvoice","w0","s0log","noctave","s0","verbose"), | |
222 | envir=environment()) | |
492cd9e7 BA |
223 | } |
224 | ||
24ed5d83 BA |
225 | # list of CWT from synchrones |
226 | # TODO: fit in RAM, OK? If not, 2 options: serialize, compute individual distances | |
492cd9e7 BA |
227 | Xcwt4 <- |
228 | if (parll) | |
229 | parallel::parLapply(cl, seq_len(n), computeCWT) | |
230 | else | |
231 | lapply(seq_len(n), computeCWT) | |
232 | ||
233 | if (parll) | |
234 | parallel::stopCluster(cl) | |
235 | ||
236 | Xwer_dist <- bigmemory::big.matrix(nrow=n, ncol=n, type="double") | |
db6fc17d | 237 | fcoefs = rep(1/3, 3) #moving average on 3 values (TODO: very slow! correct?!) |
492cd9e7 BA |
238 | if (verbose) |
239 | cat("*** Compute WER distances from CWT\n") | |
240 | ||
24ed5d83 BA |
241 | #TODO: computeDistances(i,j), et répartir les n(n-1)/2 couples d'indices |
242 | #là c'est trop déséquilibré | |
243 | ||
492cd9e7 | 244 | computeDistancesLineI = function(i) |
1c6f223e | 245 | { |
492cd9e7 BA |
246 | if (verbose) |
247 | cat(paste(" Line ",i,"\n", sep="")) | |
db6fc17d | 248 | for (j in (i+1):n) |
d03c0621 | 249 | { |
492cd9e7 | 250 | #TODO: 'circular=TRUE' is wrong, should just take values on the sides; to rewrite in C |
db6fc17d BA |
251 | num <- filter(Mod(Xcwt4[[i]] * Conj(Xcwt4[[j]])), fcoefs, circular=TRUE) |
252 | WX <- filter(Mod(Xcwt4[[i]] * Conj(Xcwt4[[i]])), fcoefs, circular=TRUE) | |
253 | WY <- filter(Mod(Xcwt4[[j]] * Conj(Xcwt4[[j]])), fcoefs, circular=TRUE) | |
24ed5d83 | 254 | wer2 <- sum(colSums(num)^2) / sum( sum(colSums(WX) * colSums(WY)) ) |
492cd9e7 BA |
255 | if (parll) |
256 | synchronicity::lock(m) | |
db6fc17d BA |
257 | Xwer_dist[i,j] <- sqrt(delta * ncol(Xcwt4[[1]]) * (1 - wer2)) |
258 | Xwer_dist[j,i] <- Xwer_dist[i,j] | |
492cd9e7 BA |
259 | if (parll) |
260 | synchronicity::unlock(m) | |
261 | } | |
262 | Xwer_dist[i,i] = 0. | |
263 | } | |
264 | ||
265 | parll = (requireNamespace("synchronicity",quietly=TRUE) | |
266 | && parll && Sys.info()['sysname'] != "Windows") | |
267 | if (parll) | |
268 | m <- synchronicity::boost.mutex() | |
269 | ||
c45fd663 | 270 | ignored <- |
492cd9e7 | 271 | if (parll) |
c45fd663 BA |
272 | { |
273 | parallel::mclapply(seq_len(n-1), computeDistancesLineI, | |
274 | mc.cores=ncores_clust, mc.allow.recursive=FALSE) | |
275 | } | |
492cd9e7 | 276 | else |
c45fd663 | 277 | lapply(seq_len(n-1), computeDistancesLineI) |
492cd9e7 | 278 | Xwer_dist[n,n] = 0. |
24ed5d83 | 279 | Xwer_dist |
492cd9e7 BA |
280 | } |
281 | ||
282 | # Helper function to divide indices into balanced sets | |
283 | .spreadIndices = function(indices, nb_per_chunk) | |
284 | { | |
285 | L = length(indices) | |
286 | nb_workers = floor( L / nb_per_chunk ) | |
287 | if (nb_workers == 0) | |
288 | { | |
289 | # L < nb_series_per_chunk, simple case | |
290 | indices_workers = list(indices) | |
291 | } | |
292 | else | |
293 | { | |
294 | indices_workers = lapply( seq_len(nb_workers), function(i) | |
295 | indices[(nb_per_chunk*(i-1)+1):(nb_per_chunk*i)] ) | |
296 | # Spread the remaining load among the workers | |
297 | rem = L %% nb_per_chunk | |
298 | while (rem > 0) | |
299 | { | |
300 | index = rem%%nb_workers + 1 | |
301 | indices_workers[[index]] = c(indices_workers[[index]], indices[L-rem+1]) | |
302 | rem = rem - 1 | |
d03c0621 | 303 | } |
1c6f223e | 304 | } |
492cd9e7 | 305 | indices_workers |
1c6f223e | 306 | } |