Commit | Line | Data |
---|---|---|
7f0781b7 | 1 | #' @include defaults.R |
3dcbfeef | 2 | |
7f0781b7 BA |
3 | #' @title Cluster power curves with PAM in parallel |
4 | #' | |
5 | #' @description Groups electricity power curves (or any series of similar nature) by applying PAM | |
cea14f3a | 6 | #' algorithm in parallel to chunks of size \code{nb_series_per_chunk} |
7f0781b7 BA |
7 | #' |
8 | #' @param data Access to the data, which can be of one of the three following types: | |
9 | #' \itemize{ | |
10 | #' \item data.frame: each line contains its ID in the first cell, and all values after | |
11 | #' \item connection: any R connection object (e.g. a file) providing lines as described above | |
12 | #' \item function: a custom way to retrieve the curves; it has two arguments: the start index | |
13 | #' (start) and number of curves (n); see example in package vignette. | |
14 | #' } | |
15 | #' @param K Number of clusters | |
cea14f3a BA |
16 | #' @param nb_series_per_chunk (Maximum) number of series in each group |
17 | #' @param min_series_per_chunk Minimum number of series in each group | |
7f0781b7 BA |
18 | #' @param writeTmp Function to write temporary wavelets coefficients (+ identifiers); |
19 | #' see defaults in defaults.R | |
20 | #' @param readTmp Function to read temporary wavelets coefficients (see defaults.R) | |
21 | #' @param WER "end" to apply stage 2 after stage 1 has iterated and finished, or "mix" | |
22 | #' to apply it after every stage 1 | |
23 | #' @param ncores number of parallel processes; if NULL, use parallel::detectCores() | |
24 | #' | |
25 | #' @return A data.frame of the final medoids curves (identifiers + values) | |
cea14f3a BA |
26 | epclust = function(data, K, nb_series_per_chunk, min_series_per_chunk=10*K, |
27 | writeTmp=defaultWriteTmp, readTmp=defaultReadTmp, WER="end", ncores=NULL) | |
ac1d4231 | 28 | { |
7f0781b7 BA |
29 | #TODO: setRefClass(...) to avoid copy data: |
30 | #http://stackoverflow.com/questions/2603184/r-pass-by-reference | |
ac1d4231 | 31 | |
7f0781b7 BA |
32 | #0) check arguments |
33 | if (!is.data.frame(data) && !is.function(data)) | |
6ecf5c2d BA |
34 | tryCatch( |
35 | { | |
36 | if (is.character(data)) | |
37 | { | |
cea14f3a | 38 | data_con = file(data, open="r") |
6ecf5c2d BA |
39 | } else if (!isOpen(data)) |
40 | { | |
41 | open(data) | |
cea14f3a | 42 | data_con = data |
6ecf5c2d BA |
43 | } |
44 | }, | |
7f0781b7 BA |
45 | error="data should be a data.frame, a function or a valid connection") |
46 | if (!is.integer(K) || K < 2) | |
47 | stop("K should be an integer greater or equal to 2") | |
cea14f3a BA |
48 | if (!is.integer(nb_series_per_chunk) || nb_series_per_chunk < K) |
49 | stop("nb_series_per_chunk should be an integer greater or equal to K") | |
7f0781b7 BA |
50 | if (!is.function(writeTmp) || !is.function(readTmp)) |
51 | stop("read/writeTmp should be functional (see defaults.R)") | |
52 | if (WER!="end" && WER!="mix") | |
53 | stop("WER takes values in {'end','mix'}") | |
54 | #concerning ncores, any non-integer type will be treated as "use parallel:detectCores()" | |
ac1d4231 | 55 | |
3d061515 | 56 | #1) acquire data (process curves, get as coeffs) |
7f0781b7 | 57 | index = 1 |
cea14f3a | 58 | nb_curves = 0 |
6ecf5c2d | 59 | repeat |
ac1d4231 | 60 | { |
cea14f3a | 61 | coeffs_chunk = NULL |
7f0781b7 | 62 | if (is.data.frame(data)) |
3dcbfeef | 63 | { |
7f0781b7 | 64 | #full data matrix |
b9f1c0c7 BA |
65 | if (index < nrow(data)) |
66 | { | |
cea14f3a BA |
67 | coeffs_chunk = curvesToCoeffs( |
68 | data[index:(min(index+nb_series_per_chunk-1,nrow(data))),]) | |
b9f1c0c7 | 69 | } |
7f0781b7 BA |
70 | } else if (is.function(data)) |
71 | { | |
72 | #custom user function to retrieve next n curves, probably to read from DB | |
cea14f3a | 73 | coeffs_chunk = curvesToCoeffs( data(index, nb_series_per_chunk) ) |
7f0781b7 BA |
74 | } else |
75 | { | |
76 | #incremental connection | |
77 | #TODO: find a better way to parse than using a temp file | |
cea14f3a | 78 | ascii_lines = readLines(data_con, nb_series_per_chunk) |
b9f1c0c7 BA |
79 | if (length(ascii_lines > 0)) |
80 | { | |
cea14f3a BA |
81 | series_chunk_file = ".tmp/series_chunk" |
82 | writeLines(ascii_lines, series_chunk_file) | |
83 | coeffs_chunk = curvesToCoeffs( read.csv(series_chunk_file) ) | |
b9f1c0c7 | 84 | } |
3dcbfeef | 85 | } |
cea14f3a BA |
86 | if (is.null(coeffs_chunk)) |
87 | break | |
88 | writeTmp(coeffs_chunk) | |
89 | nb_curves = nb_curves + nrow(coeffs_chunk) | |
90 | index = index + nb_series_per_chunk | |
8e6accca | 91 | } |
cea14f3a BA |
92 | if (exists(data_con)) |
93 | close(data_con) | |
94 | if (nb_curves < min_series_per_chunk) | |
95 | stop("Not enough data: less rows than min_series_per_chunk!") | |
8e6accca | 96 | |
cea14f3a | 97 | #2) process coeffs (by nb_series_per_chunk) and cluster them in parallel |
8e6accca | 98 | library(parallel) |
7f0781b7 | 99 | ncores = ifelse(is.integer(ncores), ncores, parallel::detectCores()) |
8e6accca | 100 | cl = parallel::makeCluster(ncores) |
7f0781b7 | 101 | parallel::clusterExport(cl=cl, varlist=c("X", "Y", "K", "p"), envir=environment()) |
6ecf5c2d | 102 | library(cluster) |
6ecf5c2d | 103 | #TODO: be careful of writing to a new temp file, then flush initial one, then re-use it... |
8e6accca BA |
104 | repeat |
105 | { | |
cea14f3a BA |
106 | #while there is jobs to do (i.e. size of tmp "file" is greater than nb_series_per_chunk) |
107 | nb_workers = nb_curves %/% nb_series_per_chunk | |
108 | indices = list() | |
109 | #incides[[i]] == (start_index,number_of_elements) | |
110 | for (i in 1:nb_workers) | |
111 | indices[[i]] = c(nb_series_per_chunk*(i-1)+1, nb_series_per_chunk) | |
112 | remainder = nb_curves %% nb_series_per_chunk | |
113 | if (remainder >= min_series_per_chunk) | |
114 | { | |
115 | nb_workers = nb_workers + 1 | |
116 | indices[[nb_workers]] = c(nb_curves-remainder+1, nb_curves) | |
117 | } else if (remainder > 0) | |
118 | { | |
119 | #spread the load among other workers | |
120 | ||
121 | } | |
122 | li = parallel::parLapply(cl, indices, processChunk, WER=="mix") | |
8e6accca | 123 | #C) flush tmp file (current parallel processes will write in it) |
8e6accca | 124 | } |
7f0781b7 | 125 | parallel::stopCluster(cl) |
3d061515 | 126 | |
8e6accca | 127 | #3) readTmp last results, apply PAM on it, and return medoids + identifiers |
cea14f3a BA |
128 | final_coeffs = readTmp(1, nb_series_per_chunk) |
129 | if (nrow(final_coeffs) == K) | |
130 | { | |
131 | return ( list( medoids=coeffsToCurves(final_coeffs[,2:ncol(final_coeffs)]), | |
132 | ids=final_coeffs[,1] ) ) | |
133 | } | |
134 | pam_output = getClusters(as.matrix(final_coeffs[,2:ncol(final_coeffs)]), K) | |
135 | medoids = coeffsToCurves(pam_output$medoids) | |
136 | ids = final_coeffs[,1] [pam_output$ranks] | |
137 | return (list(medoids=medoids, ids=ids)) | |
ac1d4231 | 138 | |
8e6accca BA |
139 | #4) apply stage 2 (in parallel ? inside task 2) ?) |
140 | if (WER == "end") | |
141 | { | |
142 | #from center curves, apply stage 2... | |
143 | } | |
ac1d4231 | 144 | } |
cea14f3a BA |
145 | |
146 | processChunk = function(indice, WER) | |
147 | { | |
148 | #1) retrieve data | |
149 | #2) cluster | |
150 | #3) WER (optional) | |
151 | } |