--- /dev/null
+#include "TimeSeries/serialize.h"
+#define __STDC_FORMAT_MACROS
+#include <inttypes.h>
+#include <math.h>
+#include "Util/types.h"
+#include "Util/utils.h"
+#include <cgds/Vector.h>
+#include <string.h>
+
+// parse a line into two integers (ID, raw power)
+static void scan_line(FILE* ifile, int posID, uint32_t* ID, int posPower, float* rawPower)
+{
+ char nextChar;
+ int position = 1;
+ while (1)
+ {
+ if (position == posID)
+ {
+ int64_t ID_on64bits;
+ nextChar = readInt(ifile, &ID_on64bits);
+ *ID = (uint32_t)ID_on64bits;
+ }
+ else if (position == posPower)
+ {
+ float power;
+ nextChar = readReal(ifile, &power);
+ *rawPower = (float) power;
+ }
+ else
+ //erase the comma (and skip field then)
+ nextChar = fgetc(ifile);
+
+ //continue until next comma (or line end or file end)
+ while (!feof(ifile) && nextChar != '\n' && nextChar != '\r' && nextChar != ',')
+ nextChar = fgetc(ifile);
+ position++;
+
+ if (feof(ifile) || nextChar == '\n' || nextChar == '\r')
+ {
+ // skip all potential line feeds
+ while (!feof(ifile) && nextChar == '\n' || nextChar == '\r')
+ nextChar = fgetc(ifile);
+ if (!feof(ifile))
+ ungetc(nextChar, ifile);
+ break;
+ }
+ }
+}
+
+//main job: parse a text file into a binary compressed version
+//TODO [long term]: adapt to distributed files/DB, maybe to distributed binary chunks
+void serialize_byCols(const char* ifileName, const char* ofileName, uint32_t nbItems)
+{
+ // use the header to know positions of ID and rawPower
+ FILE* ifile = fopen(ifileName, "r");
+ uint32_t headerShift = 0;
+ char curChar;
+ Vector* header = vector_new(char);
+ do
+ {
+ curChar = fgetc(ifile);
+ headerShift++;
+ if (curChar == '\n' || curChar == '\r')
+ {
+ //flush all potential other line feeds
+ while (curChar == '\n' || curChar == '\r')
+ curChar = fgetc(ifile);
+ ungetc(curChar, ifile);
+ break;
+ }
+ vector_push(header, curChar);
+ }
+ while (1);
+ char* headerString = (char*)malloc((vector_size(header) + 1)*sizeof(char));
+ VectorIterator* it = vector_get_iterator(header);
+ int index = 0;
+ while (vectorI_has_data(it))
+ {
+ vectorI_get(it, headerString[index]);
+ vectorI_move_next(it);
+ index++;
+ }
+ vectorI_destroy(it);
+ headerString[index] = 0;
+ vector_destroy(header);
+ int position = 1, posID = 0, posPower = 0;
+ char* columnDescriptor = strtok(headerString, ",");
+ while (columnDescriptor != NULL)
+ {
+ if (!strcmp(columnDescriptor,"FK_CCU_ID") || !strcmp(columnDescriptor,"fk_ccu_id"))
+ posID = position;
+ else if (!strcmp(columnDescriptor,"CPP_PUISSANCE_BRUTE"))
+ posPower = position;
+ position++;
+ columnDescriptor = strtok(NULL, ",");
+ }
+ free(headerString);
+
+ //estimate tsLength with a scan of the 3 first series
+ uint32_t ID=0, lastID=0, refTsLength=0;
+ float rawPower = 0.0;
+ scan_line(ifile, posID, &ID, posPower, &rawPower);
+ //'sl' = sample lengths (short because a lot of comparisons then)
+ uint32_t* sl = (uint32_t*) calloc(3, sizeof(uint32_t));
+ for (int i=0; i<3; i++)
+ {
+ lastID = ID;
+ while (ID == lastID)
+ {
+ sl[i]++;
+ scan_line(ifile, posID, &ID, posPower, &rawPower);
+ }
+ }
+ if (sl[1] <= sl[0] <= sl[2] || sl[2] <= sl[0] <= sl[1])
+ refTsLength = sl[0];
+ else if (sl[0] <= sl[1] <= sl[2] || sl[2] <= sl[1] <= sl[0])
+ refTsLength = sl[1];
+ else
+ refTsLength = sl[2];
+ free(sl);
+ //go back at the beginning of the first series (ready to read '\n'...)
+ fseek(ifile, headerShift-1, SEEK_SET);
+
+ // output file to write time-series sequentially, binary format.
+ // Format: for each series, ID is stored on 4 bytes (unsigned integer32). Then,
+ // (<rawPower>)+ follow, with rawPower stored as a float
+ FILE* ofile = fopen(ofileName, "wb");
+
+ // leave space to write the number of series (32bits), and their length in bytes (32bits)
+ for (int i = 0; i < 8; i++)
+ fputc(0, ofile);
+
+ // process one client (ID in first column) at a time
+ uint64_t processedLines = 0; //execution trace
+ uint32_t seriesCount=0, skippedSeriesCount=0, tsLength=0;
+ uint32_t mismatchLengthCount=0;
+ Byte tsBuffer[4+4*refTsLength];
+ lastID = 0;
+ while (!feof(ifile))
+ {
+ // next element to read always start with a digit
+ do
+ curChar = fgetc(ifile);
+ while (!feof(ifile) && (curChar < '0' || curChar > '9'));
+ if (feof(ifile))
+ break;
+ ungetc(curChar, ifile);
+
+ // read line
+ scan_line(ifile, posID, &ID, posPower, &rawPower);
+ if (ID != lastID)
+ {
+ //just starting a new time-series: must process the last one (if there is a last one !)
+ if (lastID > 0)
+ {
+ if (tsLength == refTsLength)
+ {
+ seriesCount++;
+ fwrite(tsBuffer, 4+4*tsLength, 1, ofile);
+ if (nbItems > 0 && seriesCount >= nbItems)
+ break;
+ }
+ //if something wrong happened, skip series
+ else
+ {
+ skippedSeriesCount++;
+ if (tsLength != refTsLength)
+ mismatchLengthCount++;
+ }
+ }
+
+ // ID for the new series is printed only once:
+ write_int(ID, tsBuffer);
+ // reinitialize flags
+ tsLength = 0;
+ lastID = ID;
+ }
+
+ //We cannot write more than refTsLength bytes
+ if (tsLength < refTsLength)
+ write_real(rawPower, tsBuffer + 4+4*tsLength);
+ tsLength++;
+
+ if ((++processedLines) % 1000000 == 0)
+ fprintf(stdout,"Processed %"PRIu64" lines\n", processedLines);
+ }
+
+ if (tsLength == refTsLength && (nbItems <= 0 || seriesCount < nbItems))
+ {
+ // flush last time-series if all conditions are met
+ fwrite(tsBuffer, 4+4*tsLength, 1, ofile);
+ seriesCount++;
+ }
+ else if (nbItems <= 0 || seriesCount < nbItems)
+ {
+ if (tsLength != refTsLength)
+ mismatchLengthCount++;
+ }
+
+ // write lines count and size of a time-series in bytes
+ Byte intBuffer[4];
+ fseek(ofile, 0, SEEK_SET);
+ write_int(seriesCount, intBuffer);
+ fwrite(intBuffer, 1, 4, ofile);
+ // re-express tsLength in bytes (not forgetting the ID))
+ write_int(4 + 4 * refTsLength, intBuffer);
+ fwrite(intBuffer, 1, 4, ofile);
+
+ // finally print some statistics
+ if (seriesCount < nbItems)
+ fprintf(stdout,"Warning: only %u series retrieved.\n",seriesCount);
+ fprintf(stdout,"%u mismatch series lengths.\n",mismatchLengthCount);
+
+ fclose(ifile);
+ fclose(ofile);
+}
+
+//serialize from usual 'by-row' data
+void serialize_byRows(const char* ifileName, const char* ofileName, uint32_t nbItems)
+{
+ FILE* ifile = fopen(ifileName, "r");
+ // first scan to know dimensions
+ uint32_t nbValues = 0; //ID (or, more useful, real class number) comes first
+ char curChar = ' ';
+ while (curChar != '\n' && curChar != '\r')
+ {
+ curChar = fgetc(ifile);
+ if (curChar == ',')
+ {
+ nbValues++;
+ //skip potential consecutive commas (could be hard to spot)
+ //TODO(...): should be 'NA' in R, thus an error (we don't handle NAs)...
+ while (curChar == ',')
+ curChar = fgetc(ifile);
+ ungetc(curChar, ifile);
+ }
+ }
+ while (curChar == '\n' || curChar == '\r')
+ curChar = fgetc(ifile);
+ ungetc(curChar, ifile);
+ uint32_t nbSeries = 1; //we already read 1st line
+ while (!feof(ifile))
+ {
+ if ((curChar = fgetc(ifile)) == '\n')
+ nbSeries++;
+ }
+ fseek(ifile, 0, SEEK_SET);
+
+ //write meta info first
+ uint32_t tsLength = 4*nbValues+4;
+ FILE* ofile = fopen(ofileName, "wb");
+ Byte intBuffer[4];
+ write_int(nbSeries, intBuffer);
+ fwrite(intBuffer, 1, 4, ofile);
+ write_int(tsLength, intBuffer);
+ fwrite(intBuffer, 1, 4, ofile);
+ float rawPower;
+ int64_t ID;
+
+ for (uint32_t i=0; i<nbSeries; i++)
+ {
+ //skip potential line feeds before next line
+ while (curChar == '\n' || curChar == '\r')
+ curChar = fgetc(ifile);
+ ungetc(curChar, ifile);
+ curChar = readInt(ifile, &ID);
+ write_int((uint32_t)ID, intBuffer);
+ fwrite(intBuffer, 1, 4, ofile);
+ while (curChar == ',')
+ curChar = fgetc(ifile);
+ ungetc(curChar, ifile);
+ for (uint32_t j=0; j<nbValues; j++)
+ {
+ curChar = readReal(ifile, &rawPower);
+ write_real(rawPower, intBuffer);
+ fwrite(intBuffer, 1, 4, ofile);
+ while (curChar == ',')
+ curChar = fgetc(ifile);
+ ungetc(curChar, ifile);
+ }
+ }
+ fclose(ifile);
+ fclose(ofile);
+}
projects / epclust.git / blobdiff