epclust/R/de_serialize.R

   1 #' @name de_serialize
   2 #' @rdname de_serialize
   3 #' @aliases binarize binarizeTransform getDataInFile
   4 #'
   5 #' @title (De)Serialization of a [big]matrix or data stream
   6 #'
   7 #' @description \code{binarize()} serializes a matrix or CSV file with minimal overhead,
   8 #'   into a binary file. \code{getDataInFile()} achieves the inverse task: she retrieves
   9 #'   (ASCII) data rows from indices in the binary file. Finally,
  10 #'   \code{binarizeTransform()} serialize transformations of all data chunks; to use it,
  11 #'   a data-retrieval function must be provided, thus \code{binarize} will most likely be
  12 #'   used first (and then a function defined to seek in generated binary file)
  13 #'
  14 #' @param data_ascii Either a matrix or CSV file, with items in rows
  15 #' @param indices Indices of the lines to retrieve
  16 #' @param data_bin_file Name of binary file on output (\code{binarize})
  17 #'   or intput (\code{getDataInFile})
  18 #' @param nb_per_chunk Number of lines to process in one batch
  19 #' @inheritParams claws
  20 #' @param getData Function to retrieve data chunks
  21 #' @param transform Transformation function to apply on data chunks
  22 #'
  23 #' @return For \code{getDataInFile()}, the matrix with rows corresponding to the
  24 #'   requested indices. \code{binarizeTransform} returns the number of processed lines.
  25 #'   \code{binarize} is designed to serialize in several calls, thus returns nothing.
  26 NULL
  27
  28 #' @rdname de_serialize
  29 #' @export
  30 binarize = function(data_ascii, data_bin_file, nb_per_chunk,
  31     sep=",", nbytes=4, endian=.Platform$endian)
  32 {
  33     if (is.character(data_ascii))
  34         data_ascii = file(data_ascii, open="r")
  35     else if (methods::is(data_ascii,"connection") && !isOpen(data_ascii))
  36         open(data_ascii)
  37     is_matrix = !methods::is(data_ascii,"connection")
  38
  39     first_write = (!file.exists(data_bin_file) || file.info(data_bin_file)$size == 0)
  40     data_bin = file(data_bin_file, open=ifelse(first_write,"wb","ab"))
  41
  42     #write data length on first call
  43     if (first_write)
  44     {
  45         #number of items always on 8 bytes
  46         writeBin(0L, data_bin, size=8, endian=endian)
  47         if (is_matrix)
  48             data_length = nrow(data_ascii)
  49         else #connection
  50         {
  51             data_line = scan(data_ascii, double(), sep=sep, nlines=1, quiet=TRUE)
  52             writeBin(data_line, data_bin, size=nbytes, endian=endian)
  53             data_length = length(data_line)
  54         }
  55     }
  56
  57     if (is_matrix)
  58         index = 1
  59     repeat
  60     {
  61         if (is_matrix)
  62         {
  63             data_chunk =
  64                 if (index <= ncol(data_ascii))
  65                     as.double(data_ascii[,index:min(ncol(data_ascii),index+nb_per_chunk-1)])
  66                 else
  67                     double(0)
  68             index = index + nb_per_chunk
  69         }
  70         else
  71             data_chunk = scan(data_ascii, double(), sep=sep, nlines=nb_per_chunk, quiet=TRUE)
  72         if (length(data_chunk)==0)
  73             break
  74         writeBin(data_chunk, data_bin, size=nbytes, endian=endian)
  75     }
  76
  77     if (first_write)
  78     {
  79         # Write data_length, = (file_size-1) / (nbytes*nbWritten) at offset 0 in data_bin
  80         ignored = seek(data_bin, 0)
  81         writeBin(data_length, data_bin, size=8, endian=endian)
  82     }
  83     close(data_bin)
  84
  85     if ( ! is_matrix )
  86         close(data_ascii)
  87 }
  88
  89 #' @rdname de_serialize
  90 #' @export
  91 binarizeTransform = function(getData, transform, data_bin_file, nb_per_chunk,
  92     nbytes=4, endian=.Platform$endian)
  93 {
  94     nb_items = 0
  95     index = 1
  96     repeat
  97     {
  98         data_chunk = getData((index-1)+seq_len(nb_per_chunk))
  99         if (is.null(data_chunk))
 100             break
 101         transformed_chunk = transform(data_chunk)
 102         binarize(transformed_chunk, data_bin_file, nb_per_chunk, ",", nbytes, endian)
 103         index = index + nb_per_chunk
 104         nb_items = nb_items + nrow(data_chunk)
 105     }
 106     nb_items
 107 }
 108
 109 #' @rdname de_serialize
 110 #' @export
 111 getDataInFile = function(indices, data_bin_file, nbytes=4, endian=.Platform$endian)
 112 {
 113     data_bin = file(data_bin_file, "rb")
 114     data_size = file.info(data_bin_file)$size
 115     data_length = readBin(data_bin, "integer", n=1, size=8, endian=endian)
 116     data_ascii = do.call( cbind, lapply( indices, function(i) {
 117         offset = 8+(i-1)*data_length*nbytes
 118         if (offset > data_size)
 119             return (NULL)
 120         ignored = seek(data_bin, offset)
 121         readBin(data_bin, "double", n=data_length, size=nbytes, endian=endian)
 122     } ) )
 123     close(data_bin)
 124     data_ascii
 125 }