| 1 | #' @title Cluster power curves with PAM in parallel |
| 2 | #' |
| 3 | #' @description Groups electricity power curves (or any series of similar nature) by applying PAM |
| 4 | #' algorithm in parallel to chunks of size \code{nb_series_per_chunk} |
| 5 | #' |
| 6 | #' @param data Access to the data, which can be of one of the three following types: |
| 7 | #' \itemize{ |
| 8 | #' \item data.frame: each line contains its ID in the first cell, and all values after |
| 9 | #' \item connection: any R connection object (e.g. a file) providing lines as described above |
| 10 | #' \item function: a custom way to retrieve the curves; it has two arguments: the ranks to be |
| 11 | #' retrieved, and the IDs - at least one of them must be present (priority: ranks). |
| 12 | #' } |
| 13 | #' @param K1 Number of super-consumers to be found after stage 1 (K1 << N) |
| 14 | #' @param K2 Number of clusters to be found after stage 2 (K2 << K1) |
| 15 | #' @param ntasks Number of tasks (parallel iterations to obtain K1 medoids); default: 1. |
| 16 | #' Note: ntasks << N, so that N is "roughly divisible" by N (number of series) |
| 17 | #' @param nb_series_per_chunk (Maximum) number of series in each group, inside a task |
| 18 | #' @param min_series_per_chunk Minimum number of series in each group |
| 19 | #' @param wf Wavelet transform filter; see ?wt.filter. Default: haar |
| 20 | #' @param WER "end" to apply stage 2 after stage 1 has iterated and finished, or "mix" |
| 21 | #' to apply it after every stage 1 |
| 22 | #' @param ncores_tasks "MPI" number of parallel tasks (1 to disable: sequential tasks) |
| 23 | #' @param ncores_clust "OpenMP" number of parallel clusterings in one task |
| 24 | #' @param random Randomize chunks repartition |
| 25 | #' |
| 26 | #' @return A data.frame of the final medoids curves (identifiers + values) |
| 27 | #' |
| 28 | #' @examples |
| 29 | #' getData = function(start, n) { |
| 30 | #' con = dbConnect(drv = RSQLite::SQLite(), dbname = "mydata.sqlite") |
| 31 | #' df = dbGetQuery(con, paste( |
| 32 | #' "SELECT * FROM times_values GROUP BY id OFFSET ",start, |
| 33 | #' "LIMIT ", n, " ORDER BY date", sep="")) |
| 34 | #' return (df) |
| 35 | #' } |
| 36 | #' #TODO: 3 examples, data.frame / binary file / DB sqLite |
| 37 | #' + sampleCurves : wavBootstrap de package wmtsa |
| 38 | #' cl = epclust(getData, K1=200, K2=15, ntasks=1000, nb_series_per_chunk=5000, WER="mix") |
| 39 | #' @export |
| 40 | epclust = function(data, K1, K2, ntasks=1, nb_series_per_chunk=50*K1, min_series_per_chunk=5*K1, |
| 41 | wf="haar", WER="end", ncores_tasks=1, ncores_clust=4, random=TRUE) |
| 42 | { |
| 43 | # Check arguments |
| 44 | if (!is.data.frame(data) && !is.function(data)) |
| 45 | { |
| 46 | tryCatch( |
| 47 | { |
| 48 | if (is.character(data)) |
| 49 | data_con = file(data, open="r") |
| 50 | else if (!isOpen(data)) |
| 51 | { |
| 52 | open(data) |
| 53 | data_con = data |
| 54 | } |
| 55 | }, |
| 56 | error=function(e) "data should be a data.frame, a function or a valid connection" |
| 57 | ) |
| 58 | } |
| 59 | K1 = toInteger(K1, function(x) x>=2) |
| 60 | K2 = toInteger(K2, function(x) x>=2) |
| 61 | ntasks = toInteger(ntasks, function(x) x>=1) |
| 62 | nb_series_per_chunk = toInteger(nb_series_per_chunk, function(x) x>=K1) |
| 63 | min_series_per_chunk = toInteger(K1, function(x) x>=K1 && x<=nb_series_per_chunk) |
| 64 | ncores_tasks = toInteger(ncores_tasks, function(x) x>=1) |
| 65 | ncores_clust = toInteger(ncores_clust, function(x) x>=1) |
| 66 | if (WER!="end" && WER!="mix") |
| 67 | stop("WER takes values in {'end','mix'}") |
| 68 | |
| 69 | # Serialize all wavelets coefficients (+ IDs) onto a file |
| 70 | unlink(".coeffs") |
| 71 | index = 1 |
| 72 | nb_curves = 0 |
| 73 | nb_coeffs = NA |
| 74 | repeat |
| 75 | { |
| 76 | coeffs_chunk = computeCoeffs(data, index, nb_series_per_chunk, wf) |
| 77 | if (is.null(coeffs_chunk)) |
| 78 | break |
| 79 | writeCoeffs(coeffs_chunk) |
| 80 | index = index + nb_series_per_chunk |
| 81 | nb_curves = nb_curves + nrow(coeffs_chunk) |
| 82 | if (is.na(nb_coeffs)) |
| 83 | nb_coeffs = ncol(coeffs_chunk)-1 |
| 84 | } |
| 85 | |
| 86 | if (nb_curves < min_series_per_chunk) |
| 87 | stop("Not enough data: less rows than min_series_per_chunk!") |
| 88 | nb_series_per_task = round(nb_curves / ntasks) |
| 89 | if (nb_series_per_task < min_series_per_chunk) |
| 90 | stop("Too many tasks: less series in one task than min_series_per_chunk!") |
| 91 | |
| 92 | # Cluster coefficients in parallel (by nb_series_per_chunk) |
| 93 | indices = if (random) sample(nb_curves) else seq_len(nb_curves) |
| 94 | indices_tasks = lapply(seq_len(ntasks), function(i) { |
| 95 | upper_bound = ifelse( i<ntasks, min(nb_series_per_task*i,nb_curves), nb_curves ) |
| 96 | indices[((i-1)*nb_series_per_task+1):upper_bound] |
| 97 | }) |
| 98 | library(parallel, quietly=TRUE) |
| 99 | cl_tasks = parallel::makeCluster(ncores_tasks) |
| 100 | parallel::clusterExport(cl_tasks, |
| 101 | varlist=c("K1","K2","WER","nb_series_per_chunk","ncores_clust"),#TODO: pass also |
| 102 | #nb_coeffs...and filename (in a list... ?) |
| 103 | envir=environment()) |
| 104 | indices = parallel::parLapply(cl_tasks, indices_tasks, clusteringTask) |
| 105 | parallel::stopCluster(cl_tasks) |
| 106 | |
| 107 | # Run step1+2 step on resulting ranks |
| 108 | indices = clusterChunk(indices, K1, K2) |
| 109 | return (list("indices"=indices, "medoids"=getSeries(data, indices))) |
| 110 | } |