forked from BodyTrack/datastore
-
Notifications
You must be signed in to change notification settings - Fork 0
/
gettile.cpp
304 lines (265 loc) · 11.3 KB
/
gettile.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
// C++
#include <iostream>
#include <set>
#include <string>
#include <vector>
// C
#include <math.h>
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
// Local
#include "Binrec.h"
#include "Channel.h"
#include "fft.h"
#include "FilesystemKVS.h"
#include "ImportBT.h"
#include "Log.h"
#include "utils.h"
void usage()
{
std::cerr << "Usage:\n";
std::cerr << "gettile store.kvs UID devicenickname.channel level offset\n";
#if FFT_SUPPORT
std::cerr << " If the string '.DFT' is appended to the channel name, the discrete\n";
std::cerr << " Fourier transform of the data is returned instead\n";
#endif /* FFT_SUPPORT */
std::cerr << "Exiting...\n";
exit(1);
}
template <typename T>
void read_tile_samples(KVS &store, int uid, std::string full_channel_name, TileIndex requested_index, TileIndex client_tile_index, std::vector<DataSample<T> > &samples, bool &binned)
{
Channel ch(store, uid, full_channel_name);
Tile tile;
TileIndex actual_index;
bool success = ch.read_tile_or_closest_ancestor(requested_index, actual_index, tile);
if (!success) {
log_f("gettile: no tile found for %s", requested_index.to_string().c_str());
} else {
log_f("gettile: requested %s: found %s", requested_index.to_string().c_str(), actual_index.to_string().c_str());
for (unsigned i = 0; i < tile.get_samples<T>().size(); i++) {
DataSample<T> &sample=tile.get_samples<T>()[i];
if (client_tile_index.contains_time(sample.time)) samples.push_back(sample);
}
}
if (samples.size() <= 512) {
binned = false;
} else {
// Bin
binned = true;
std::vector<DataAccumulator<T> > bins(512);
for (unsigned i = 0; i < samples.size(); i++) {
DataSample<T> &sample=samples[i];
bins[(int)floor(client_tile_index.position(sample.time)*512)] += sample;
}
samples.clear();
for (unsigned i = 0; i < bins.size(); i++) {
if (bins[i].weight > 0) samples.push_back(bins[i].get_sample());
}
}
}
struct GraphSample {
double time;
bool has_value;
double value;
double stddev;
double weight;
bool has_comment;
std::string comment;
GraphSample(DataSample<double> &x) : time(x.time), has_value(true), value(x.value), stddev(x.stddev), weight(x.weight),
has_comment(false), comment("") {}
GraphSample(DataSample<std::string> &x) : time(x.time), has_value(false), value(0), stddev(x.stddev), weight(x.weight),
has_comment(true), comment(x.value) {}
};
bool operator<(const GraphSample &a, const GraphSample &b) { return a.time < b.time; }
int main(int argc, char **argv)
{
long long begin_time = millitime();
char **argptr = argv+1;
if (!*argptr) usage();
std::string storename = *argptr++;
if (!*argptr) usage();
int uid = atoi(*argptr++);
set_log_prefix(string_printf("%d %d ", getpid(), uid));
if (!*argptr) usage();
std::string full_channel_name = *argptr++;
#if FFT_SUPPORT
bool writing_fft = false;
size_t fftpos = full_channel_name.rfind(".DFT");
if (fftpos != std::string::npos) {
full_channel_name = full_channel_name.substr(0, fftpos);
writing_fft = true;
}
#endif /* FFT_SUPPORT */
if (!*argptr) usage();
int tile_level = atoi(*argptr++);
if (!*argptr) usage();
long long tile_offset = atoll(*argptr++);
if (*argptr) usage();
// Desired level and offset
// Translation between tile request and tilestore:
// tile: level 0 is 512 samples in 512 seconds
// store: level 0 is 65536 samples in 1 second
// for tile level 0, we want to get store level 14, which is 65536 samples in 16384 seconds
// Levels differ by 9 between client and server
TileIndex client_tile_index = TileIndex(tile_level+9, tile_offset);
{
std::string arglist;
for (int i = 0; i < argc; i++) {
if (i) arglist += " ";
arglist += std::string("'")+argv[i]+"'";
}
log_f("gettile START: %s (time %.9f-%.9f)",
arglist.c_str(), client_tile_index.start_time(), client_tile_index.end_time());
}
FilesystemKVS store(storename.c_str());
// 5th ancestor
TileIndex requested_index = client_tile_index.parent().parent().parent().parent().parent();
std::vector<DataSample<double> > double_samples;
std::vector<DataSample<std::string> > string_samples;
std::vector<DataSample<std::string> > comments;
bool doubles_binned, strings_binned, comments_binned;
// TODO: If writing FFT, ***get more data***
// TODO: Use min_time_required and max_time_required, get max-res data
read_tile_samples(store, uid, full_channel_name, requested_index, client_tile_index, double_samples, doubles_binned);
#if FFT_SUPPORT
if (writing_fft) {
std::vector<std::vector<double> > fft, shifted;
int num_values;
windowed_fft(double_samples, requested_index, fft);
present_fft(fft, shifted, num_values);
// JSON tile to send back to the client includes some of the same
// information as a non-DFT tile
Json::Value tile(Json::objectValue);
tile["level"] = Json::Value(tile_level);
// See discussion below for reason to cast tile_offset
// from long long to double
tile["offset"] = Json::Value((double)tile_offset);
tile["num_values"] = Json::Value(num_values);
tile["dft"] = Json::Value(Json::arrayValue);
for (unsigned window_id = 0; window_id < shifted.size(); window_id++) {
Json::Value window(Json::arrayValue);
for (unsigned i = 0; i < shifted[window_id].size(); i++)
window.append(shifted[window_id][i]);
tile["dft"].append(window);
}
std::cout << Json::FastWriter().write(tile) << std::endl;
return 0;
}
#endif /* FFT_SUPPORT */
read_tile_samples(store, uid, full_channel_name, requested_index, client_tile_index, string_samples, strings_binned);
read_tile_samples(store, uid, full_channel_name+"._comment", requested_index, client_tile_index, comments, comments_binned);
string_samples.insert(string_samples.end(), comments.begin(), comments.end());
std::sort(string_samples.begin(), string_samples.end(), DataSample<std::string>::time_lessthan);
std::map<double, DataSample<double> > double_sample_map;
for (unsigned i = 0; i < double_samples.size(); i++) {
double_sample_map[double_samples[i].time] = double_samples[i]; // TODO: combine if two samples at same time?
}
std::set<double> has_string;
for (unsigned i = 0; i < string_samples.size(); i++) {
has_string.insert(string_samples[i].time);
}
std::vector<GraphSample> graph_samples;
bool has_fifth_col = string_samples.size()>0;
for (unsigned i = 0; i < string_samples.size(); i++) {
if (double_sample_map.find(string_samples[i].time) != double_sample_map.end()) {
GraphSample gs(double_sample_map[string_samples[i].time]);
gs.has_comment = true;
gs.comment = string_samples[i].value;
graph_samples.push_back(gs);
} else {
graph_samples.push_back(GraphSample(string_samples[i]));
}
}
for (unsigned i = 0; i < double_samples.size(); i++) {
if (has_string.find(double_samples[i].time) == has_string.end()) {
graph_samples.push_back(GraphSample(double_samples[i]));
}
}
std::sort(graph_samples.begin(), graph_samples.end());
double bin_width = client_tile_index.duration() / 512.0;
double line_break_threshold = bin_width * 4.0;
if (!doubles_binned && double_samples.size() > 1) {
// Find the median distance between samples
std::vector<double> spacing(double_samples.size()-1);
for (size_t i = 0; i < double_samples.size()-1; i++) {
spacing[i] = double_samples[i+1].time - double_samples[i].time;
}
std::sort(spacing.begin(), spacing.end());
double median_spacing = spacing[spacing.size()/2];
// Set line_break_threshold to larger of 4*median_spacing and 4*bin_width
line_break_threshold = std::max(line_break_threshold, median_spacing * 4);
}
if (graph_samples.size()) {
log_f("gettile: outputting %zd samples", graph_samples.size());
Json::Value tile(Json::objectValue);
tile["level"] = Json::Value(tile_level);
// An aside about offset type and precision:
// JSONCPP doesn't have a long long type; to preserve full resolution we need to convert to double here. As Javascript itself
// will read this as a double-precision value, we're not introducing a problem.
// For a detailed discussion, see https://sites.google.com/a/bodytrack.org/wiki/website/tile-coordinates-and-numeric-precision
// Irritatingly, JSONCPP wants to add ".0" to the end of floating-point numbers that don't need it. This is inconsistent
// with Javascript itself and simply introduces extra bytes to the representation
tile["offset"] = Json::Value((double)tile_offset);
tile["fields"] = Json::Value(Json::arrayValue);
tile["fields"].append(Json::Value("time"));
tile["fields"].append(Json::Value("mean"));
tile["fields"].append(Json::Value("stddev"));
tile["fields"].append(Json::Value("count"));
if (has_fifth_col) tile["fields"].append(Json::Value("comment"));
Json::Value data(Json::arrayValue);
double previous_sample_time = client_tile_index.start_time();
bool previous_had_value = true;
for (unsigned i = 0; i < graph_samples.size(); i++) {
// TODO: improve linebreak calculations:
// 1) observe channel specs line break size from database (expressed in time; some observations have long time periods and others short)
// 2) insert breaks at beginning or end of tile if needed
// 3) should client be the one to decide where line breaks are (if we give it the threshold?)
if (graph_samples[i].time - previous_sample_time > line_break_threshold ||
!graph_samples[i].has_value || !previous_had_value) {
// Insert line break, which has value -1e+308
Json::Value sample = Json::Value(Json::arrayValue);
sample.append(Json::Value(0.5*(graph_samples[i].time+previous_sample_time)));
sample.append(Json::Value(-1e308));
sample.append(Json::Value(0));
sample.append(Json::Value(0));
if (has_fifth_col) sample.append(Json::Value()); // NULL
data.append(sample);
}
previous_sample_time = graph_samples[i].time;
previous_had_value = graph_samples[i].has_value;
{
Json::Value sample = Json::Value(Json::arrayValue);
sample.append(Json::Value(graph_samples[i].time));
sample.append(Json::Value(graph_samples[i].has_value ? graph_samples[i].value : 0.0));
// TODO: fix datastore so we never see NAN crop up here!
sample.append(Json::Value(isnan(graph_samples[i].stddev) ? 0 : graph_samples[i].stddev));
sample.append(Json::Value(graph_samples[i].weight));
if (has_fifth_col) {
sample.append(graph_samples[i].has_comment ? Json::Value(graph_samples[i].comment) : Json::Value());
}
data.append(sample);
}
}
if (client_tile_index.end_time() - previous_sample_time > line_break_threshold ||
!previous_had_value) {
// Insert line break, which has value -1e+308
Json::Value sample = Json::Value(Json::arrayValue);
sample.append(Json::Value(0.5*(previous_sample_time + client_tile_index.end_time())));
sample.append(Json::Value(-1e308));
sample.append(Json::Value(0));
sample.append(Json::Value(0));
if (has_fifth_col) sample.append(Json::Value()); // NULL
data.append(sample);
}
tile["data"] = data;
tile["sample_width"] = std::max(30.0, bin_width);
printf("%s\n", rtrim(Json::FastWriter().write(tile)).c_str());
} else {
log_f("gettile: no samples");
printf("{}");
}
log_f("gettile: finished in %lld msec", millitime() - begin_time);
return 0;
}