Commit | Line | Data |
---|---|---|
3dcbfeef BA |
1 | #TODO: setRefClass... to avoid copy data !! |
2 | #http://stackoverflow.com/questions/2603184/r-pass-by-reference | |
3 | ||
4 | #fields: data (can be NULL or provided by user), coeffs (will be computed | |
5 | #con can be a character string naming a file; see readLines() | |
6 | #data can be in DB format, on one column : TODO: guess (from header, or col. length...) | |
3d061515 BA |
7 | |
8 | ||
9 | writeTmp(curves [uncompressed coeffs, limited number - nbSeriesPerChunk], last=FALSE) #if last=TRUE, close the conn | |
10 | readTmp(..., from index, n curves) #careful: connection must remain open | |
11 | #TODO: write read/write tmp reference ( on file in .tmp/ folder ... ) | |
12 | ||
8e6accca BA |
13 | #data: |
14 | #stop("Unrecognizable 'data' argument (must be numeric, functional or connection)") | |
15 | ||
16 | #WER: "end" to apply stage 2 after stage 1 iterated, "mix" (or anything else...?!) to apply it after every stage 1 | |
17 | epclust = function(data, K, nbPerChunk, WER="end", ncores=NULL, writeTmp=ref_writeTmp, readTmp=ref_readTmp) #where to put/retrieve intermediate results; if not provided, use file on disk | |
ac1d4231 BA |
18 | { |
19 | ||
20 | ||
ac1d4231 | 21 | #on input: can be data or con; data handled by writing it to file (ascii or bin ?!), |
3d061515 | 22 | #data: con or matrix or DB |
ac1d4231 | 23 | |
3d061515 BA |
24 | #1) acquire data (process curves, get as coeffs) |
25 | if (is.numeric(data)) | |
ac1d4231 BA |
26 | { |
27 | #full data matrix | |
3dcbfeef BA |
28 | index = 1 |
29 | n = nrow(data) | |
30 | while (index < n) | |
31 | { | |
3d061515 | 32 | writeTmp( getCoeffs(data) ) |
3dcbfeef BA |
33 | index = index + nbSeriesPerChunk |
34 | } | |
3d061515 BA |
35 | } else if (is.function(data)) |
36 | { | |
37 | #custom user function to retrieve next n curves, probably to read from DB | |
38 | writeTmp( getCoeffs( data(nbPerChunk) ) ) | |
39 | } else | |
ac1d4231 BA |
40 | { |
41 | #incremental connection | |
42 | #read it one by one and get coeffs until nbSeriesPerChunk | |
43 | #then launch a clustering task............ | |
8e6accca | 44 | #TODO: find a better way to parse than using a temp file |
3d061515 | 45 | ascii_lines = readLines(data, nbSeriesPerChunk) |
8e6accca | 46 | seriesChunkFile = ".tmp/seriesChunk" |
3d061515 BA |
47 | writeLines(ascii_lines, seriesChunkFile) |
48 | writeTmp( getCoeffs( read.csv(seriesChunkFile) ) ) | |
8e6accca BA |
49 | } |
50 | ||
51 | library(parallel) | |
52 | ncores = ifelse(is.numeric(ncores), ncores, parallel::detectCores()) | |
53 | cl = parallel::makeCluster(ncores) | |
54 | 115 parallel::clusterExport(cl=cl, varlist=c("X", "Y", "K", "p"), envir=environment()) | |
55 | 116 li = parallel::parLapply(cl, 1:B, getParamsAtIndex) | |
3d061515 BA |
56 | |
57 | #2) process coeffs (by nbSeriesPerChunk) and cluster in parallel (just launch async task, wait for them to complete, and re-do if necessary) | |
8e6accca BA |
58 | repeat |
59 | { | |
60 | completed = rep(FALSE, ............) | |
61 | #while there is jobs to do (i.e. size of tmp "file" is greater than nbSeriesPerChunk), | |
62 | #A) determine which tasks which processor will do (OK) | |
63 | #B) send each (sets of) tasks in parallel | |
64 | #C) flush tmp file (current parallel processes will write in it) | |
65 | #always check "complete" flag (array, as I did in MPI) to know if "slaves" finished | |
66 | } | |
67 | ||
68 | parallel::stopCluster(cl) | |
3d061515 | 69 | |
8e6accca | 70 | #3) readTmp last results, apply PAM on it, and return medoids + identifiers |
ac1d4231 | 71 | |
8e6accca BA |
72 | #4) apply stage 2 (in parallel ? inside task 2) ?) |
73 | if (WER == "end") | |
74 | { | |
75 | #from center curves, apply stage 2... | |
76 | } | |
ac1d4231 | 77 | } |
3dcbfeef BA |
78 | |
79 | getCoeffs = function(series) | |
80 | { | |
81 | #... return wavelets coeffs : compute in parallel ! | |
82 | } |