improvements
[epclust.git] / 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 = ncol(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 <= nrow(data_ascii))
65 as.double(t(data_ascii[index:min(nrow(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 #Ou t(sapply(...)) (+ rapide ?)
117 data_ascii = do.call( rbind, lapply( indices, function(i) {
118 offset = 8+(i-1)*data_length*nbytes
119 if (offset > data_size)
120 return (vector("double",0))
121 ignored = seek(data_bin, offset)
122 readBin(data_bin, "double", n=data_length, size=nbytes, endian=endian)
123 } ) )
124 close(data_bin)
125 if (ncol(data_ascii)>0) data_ascii else NULL
126 }