code/draft_R_pkg/R/main.R

   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...)
   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
  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
  18 {
  19
  20
  21     #on input: can be data or con; data handled by writing it to file (ascii or bin ?!),
  22 #data: con or matrix or DB
  23
  24     #1) acquire data (process curves, get as coeffs)
  25     if (is.numeric(data))
  26     {
  27         #full data matrix
  28         index = 1
  29         n = nrow(data)
  30         while (index < n)
  31         {
  32             writeTmp( getCoeffs(data) )
  33             index = index + nbSeriesPerChunk
  34         }
  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
  40     {
  41         #incremental connection
  42         #read it one by one and get coeffs until nbSeriesPerChunk
  43         #then launch a clustering task............
  44         #TODO: find a better way to parse than using a temp file
  45         ascii_lines = readLines(data, nbSeriesPerChunk)
  46         seriesChunkFile = ".tmp/seriesChunk"
  47         writeLines(ascii_lines, seriesChunkFile)
  48         writeTmp( getCoeffs( read.csv(seriesChunkFile) ) )
  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)
  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)
  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)
  69
  70     #3) readTmp last results, apply PAM on it, and return medoids + identifiers
  71
  72     #4) apply stage 2 (in parallel ? inside task 2) ?)
  73     if (WER == "end")
  74     {
  75         #from center curves, apply stage 2...
  76     }
  77 }
  78
  79 getCoeffs = function(series)
  80 {
  81     #... return wavelets coeffs : compute in parallel !
  82 }