diff --git a/include/villas/sample_io.h b/include/villas/sample_io.h new file mode 100644 index 000000000..4e14fe809 --- /dev/null +++ b/include/villas/sample_io.h @@ -0,0 +1,83 @@ +/** Read / write sample data in different formats. + * + * @file + * @author Steffen Vogel + * @copyright 2017, Institute for Automation of Complex Power Systems, EONERC + *********************************************************************************/ + +#pragma once + +#include +#include + +/* Forward declarations */ +struct sample; + +enum sample_io_format { + SAMPLE_IO_FORMAT_VILLAS, + SAMPLE_IO_FORMAT_JSON, + SAMPLE_IO_FORMAT_HDF5, + SAMPLE_IO_FORMAT_COMTRADE +}; + +/** These flags define the format which is used by sample_io_fscan() and sample_io_fprint(). */ +enum sample_flags { + SAMPLE_IO_NANOSECONDS = (1 << 0), + SAMPLE_IO_OFFSET = (1 << 1), + SAMPLE_IO_SEQUENCE = (1 << 2), + SAMPLE_IO_VALUES = (1 << 3), + SAMPLE_IO_ALL = 16-1 +}; + +/* Not implemented yet */ +#if 0 +struct sample_io_handle { + enum sample_io_format format; + int flags + + FILE *file; + + struct list fields; +}; + +int sample_io_init(struct sample_io *io, enum sample_io_format fmt, char *mode, int flags); + +int sample_io_destroy(struct sample_io *io); + +int sample_io_open(struct sample_io *io); + +int sample_io_close(struct sample_io *io); + +int sample_io_write(struct sample_io *io, struct sample *smps[], size_t cnt); + +int sample_io_read(struct sample_io *io, struct sample *smps[], size_t cnt); + +int sample_io_eof(struct sample_io *io); +int sample_io_rewind(struct sample_io *io); +#endif + +/* Lowlevel interface */ + +int sample_io_fprint(FILE *f, struct sample *s, enum sample_io_format fmt, int flags); + +int sample_io_fscan(FILE *f, struct sample *s, enum sample_io_format fmt, int *flags); + +/* VILLASnode human readable format */ + +int sample_io_villas_print(char *buf, size_t len, struct sample *s, int flags); + +int sample_io_villas_scan(const char *line, struct sample *s, int *fl); + +int sample_io_villas_fprint(FILE *f, struct sample *s, int flags); + +int sample_io_villas_fscan(FILE *f, struct sample *s, int *flags); + +/* JSON format */ + +int sample_io_json_pack(json_t **j, struct sample *s, int flags); + +int sample_io_json_unpack(json_t *j, struct sample *s, int *flags); + +int sample_io_json_fprint(FILE *f, struct sample *s, int flags); + +int sample_io_json_fscan(FILE *f, struct sample *s, int *flags); diff --git a/lib/Makefile.inc b/lib/Makefile.inc index f0f8ec112..bcce31a13 100644 --- a/lib/Makefile.inc +++ b/lib/Makefile.inc @@ -7,7 +7,7 @@ LIB_SRCS = $(addprefix lib/nodes/, file.c cbuilder.c) \ $(addprefix lib/, sample.c path.c node.c hook.c \ log.c utils.c super_node.c hist.c timing.c pool.c \ list.c queue.c memory.c advio.c web.c api.c \ - plugin.c node_type.c stats.c mapping.c \ + plugin.c node_type.c stats.c mapping.c sample_io.c\ ) LIB_CFLAGS = $(CFLAGS) -fPIC diff --git a/lib/sample_io.c b/lib/sample_io.c new file mode 100644 index 000000000..ccd1e871a --- /dev/null +++ b/lib/sample_io.c @@ -0,0 +1,282 @@ +/** The internal datastructure for a sample of simulation data. + * + * @author Steffen Vogel + * @copyright 2017, Institute for Automation of Complex Power Systems, EONERC + *********************************************************************************/ + +#include + +#include "sample.h" +#include "sample_io.h" +#include "timing.h" + +int sample_io_fprint(FILE *f, struct sample *s, enum sample_io_format fmt, int flags) +{ + switch (fmt) { + case SAMPLE_IO_FORMAT_VILLAS: return sample_io_villas_fprint(f, s, flags); + case SAMPLE_IO_FORMAT_JSON: return sample_io_json_fprint(f, s, flags); + default: + return -1; + } +} + +int sample_io_fscan(FILE *f, struct sample *s, enum sample_io_format fmt, int *flags) +{ + switch (fmt) { + case SAMPLE_IO_FORMAT_VILLAS: return sample_io_villas_fscan(f, s, flags); + case SAMPLE_IO_FORMAT_JSON: return sample_io_json_fscan(f, s, flags); + default: + return -1; + } +} + +int sample_io_villas_print(char *buf, size_t len, struct sample *s, int flags) +{ + size_t off = snprintf(buf, len, "%llu", (unsigned long long) s->ts.origin.tv_sec); + + if (flags & SAMPLE_IO_NANOSECONDS) + off += snprintf(buf + off, len - off, ".%09llu", (unsigned long long) s->ts.origin.tv_nsec); + + if (flags & SAMPLE_IO_OFFSET) + off += snprintf(buf + off, len - off, "%+e", time_delta(&s->ts.origin, &s->ts.received)); + + if (flags & SAMPLE_IO_SEQUENCE) + off += snprintf(buf + off, len - off, "(%u)", s->sequence); + + if (flags & SAMPLE_IO_VALUES) { + for (int i = 0; i < s->length; i++) { + switch ((s->format >> i) & 0x1) { + case SAMPLE_DATA_FORMAT_FLOAT: + off += snprintf(buf + off, len - off, "\t%.6f", s->data[i].f); + break; + case SAMPLE_DATA_FORMAT_INT: + off += snprintf(buf + off, len - off, "\t%d", s->data[i].i); + break; + } + } + } + + off += snprintf(buf + off, len - off, "\n"); + + return 0; /* trailing '\0' */ +} + +int sample_io_villas_scan(const char *line, struct sample *s, int *fl) +{ + char *end; + const char *ptr = line; + + int flags = 0; + double offset = 0; + + /* Format: Seconds.NanoSeconds+Offset(SequenceNumber) Value1 Value2 ... + * RegEx: (\d+(?:\.\d+)?)([-+]\d+(?:\.\d+)?(?:e[+-]?\d+)?)?(?:\((\d+)\))? + * + * Please note that only the seconds and at least one value are mandatory + */ + + /* Mandatory: seconds */ + s->ts.origin.tv_sec = (uint32_t) strtoul(ptr, &end, 10); + if (ptr == end) + return -2; + + /* Optional: nano seconds */ + if (*end == '.') { + ptr = end + 1; + + s->ts.origin.tv_nsec = (uint32_t) strtoul(ptr, &end, 10); + if (ptr != end) + flags |= SAMPLE_IO_NANOSECONDS; + else + return -3; + } + else + s->ts.origin.tv_nsec = 0; + + /* Optional: offset / delay */ + if (*end == '+' || *end == '-') { + ptr = end; + + offset = strtof(ptr, &end); /* offset is ignored for now */ + if (ptr != end) + flags |= SAMPLE_IO_OFFSET; + else + return -4; + } + + /* Optional: sequence */ + if (*end == '(') { + ptr = end + 1; + + s->sequence = strtoul(ptr, &end, 10); + if (ptr != end) + flags |= SAMPLE_IO_SEQUENCE; + else + return -5; + + if (*end == ')') + end++; + } + + for (ptr = end, s->length = 0; + s->length < s->capacity; + ptr = end, s->length++) { + + switch (s->format & (1 << s->length)) { + case SAMPLE_DATA_FORMAT_FLOAT: + s->data[s->length].f = strtod(ptr, &end); + break; + case SAMPLE_DATA_FORMAT_INT: + s->data[s->length].i = strtol(ptr, &end, 10); + break; + } + + if (end == ptr) /* There are no valid FP values anymore */ + break; + } + + if (s->length > 0) + flags |= SAMPLE_IO_VALUES; + + if (fl) + *fl = flags; + if (flags & SAMPLE_IO_OFFSET) { + struct timespec off = time_from_double(offset); + s->ts.received = time_add(&s->ts.origin, &off); + } + else + s->ts.received = s->ts.origin; + + return 0; +} + +int sample_io_villas_fprint(FILE *f, struct sample *s, int flags) +{ + char line[4096]; + int ret; + + ret = sample_io_villas_print(line, sizeof(line), s, flags); + if (ret) + return ret; + + fputs(line, f); + + return 0; +} + +int sample_io_villas_fscan(FILE *f, struct sample *s, int *fl) +{ + char *ptr, line[4096]; + +skip: if (fgets(line, sizeof(line), f) == NULL) + return -1; /* An error occured */ + + /* Skip whitespaces, empty and comment lines */ + for (ptr = line; isspace(*ptr); ptr++); + if (*ptr == '\0' || *ptr == '#') + goto skip; + + return sample_io_villas_scan(line, s, fl); +} + +#ifdef WITH_JANSSON +int sample_io_json_pack(json_t **j, struct sample *s, int flags) +{ + json_error_t err; + json_t *json_data = json_array(); + + for (int i = 0; i < s->length; i++) { + json_t *json_value = sample_get_data_format(s, i) + ? json_integer(s->data[i].i) + : json_real(s->data[i].f); + + json_array_append(json_data, json_value); + } + + *j = json_pack_ex(&err, 0, "{ s: { s: [ I, I ], s: [ I, I ], s: [ I, I ] }, s: I, s: o }", + "ts", + "origin", s->ts.origin.tv_sec, s->ts.origin.tv_nsec, + "received", s->ts.received.tv_sec, s->ts.received.tv_nsec, + "sent", s->ts.sent.tv_sec, s->ts.sent.tv_nsec, + "sequence", s->sequence, + "data", json_data); + + if (!*j) + return -1; + + return 0; +} + +int sample_io_json_unpack(json_t *j, struct sample *s, int *flags) +{ + int ret, i; + json_t *json_data, *json_value; + + ret = json_unpack(j, "{ s: { s: [ I, I ], s: [ I, I ], s: [ I, I ] }, s: I, s: o }", + "ts", + "origin", &s->ts.origin.tv_sec, &s->ts.origin.tv_nsec, + "received", &s->ts.received.tv_sec, &s->ts.received.tv_nsec, + "sent", &s->ts.sent.tv_sec, &s->ts.sent.tv_nsec, + "sequence", &s->sequence, + "data", &json_data); + + if (ret) + return ret; + + s->length = 0; + + json_array_foreach(json_data, i, json_value) { + switch (json_typeof(json_value)) { + case JSON_REAL: + s->data[i].f = json_real_value(json_value); + sample_set_data_format(s, i, SAMPLE_DATA_FORMAT_FLOAT); + break; + + case JSON_INTEGER: + s->data[i].f = json_integer_value(json_value); + sample_set_data_format(s, i, SAMPLE_DATA_FORMAT_INT); + break; + + default: + return -1; + } + + s->length++; + } + + return 0; +} + +int sample_io_json_fprint(FILE *f, struct sample *s, int flags) +{ + int ret; + json_t *json; + + ret = sample_io_json_pack(&json, s, flags); + if (ret) + return ret; + + ret = json_dumpf(json, f, 0); + + json_decref(json); + + return ret; +} + +int sample_io_json_fscan(FILE *f, struct sample *s, int *flags) +{ + int ret; + json_t *json; + json_error_t err; + + json = json_loadf(f, JSON_DISABLE_EOF_CHECK, &err); + if (!json) + return -1; + + ret = sample_io_json_unpack(json, s, flags); + + json_decref(json); + + return ret; +} +#endif \ No newline at end of file diff --git a/src/hook.c b/src/hook.c index 986ff490e..b740fcaed 100644 --- a/src/hook.c +++ b/src/hook.c @@ -15,6 +15,7 @@ #include #include +#include #include #include #include @@ -155,7 +156,7 @@ int main(int argc, char *argv[]) recv = 0; for (int j = 0; j < cnt && !feof(stdin); j++) { - ret = sample_fscan(stdin, hi.samples[j], NULL); + ret = sample_io_villas_fscan(stdin, samples[j], NULL); if (ret < 0) break; @@ -168,8 +169,8 @@ int main(int argc, char *argv[]) hook_read(&h, samples, &recv); hook_write(&h, samples, &recv); - for (int j = 0; j < hi.count; j++) - sample_fprint(stdout, hi.samples[j], SAMPLE_ALL); + for (int j = 0; j < recv; j++) + sample_io_villas_fprint(stdout, samples[j], SAMPLE_IO_ALL); fflush(stdout); sample_free(samples, cnt); diff --git a/src/pipe.c b/src/pipe.c index bf3fcbeab..8b2b99659 100644 --- a/src/pipe.c +++ b/src/pipe.c @@ -20,6 +20,7 @@ #include #include #include +#include #include #include "config.h" @@ -103,7 +104,7 @@ static void * send_loop(void *ctx) struct sample *s = smps[i]; int reason; -retry: reason = sample_fscan(stdin, s, NULL); +retry: reason = sample_io_villas_fscan(stdin, s, NULL); if (reason < 0) { if (feof(stdin)) { info("Reached end-of-file. Terminating..."); @@ -157,7 +158,7 @@ static void * recv_loop(void *ctx) if (s->ts.received.tv_sec == -1 || s->ts.received.tv_sec == 0) s->ts.received = now; - sample_fprint(stdout, s, SAMPLE_ALL); + sample_io_villas_fprint(stdout, s, SAMPLE_IO_ALL); fflush(stdout); } pthread_testcancel(); diff --git a/src/signal.c b/src/signal.c index 1e2121548..aa36ff739 100644 --- a/src/signal.c +++ b/src/signal.c @@ -14,6 +14,7 @@ #include #include +#include #include #include "config.h" @@ -176,7 +177,7 @@ check: if (optarg == endptr) } } - sample_fprint(stdout, s, SAMPLE_ALL & ~SAMPLE_OFFSET); + sample_io_villas_fprint(stdout, s, SAMPLE_IO_ALL & ~SAMPLE_IO_OFFSET); fflush(stdout); /* Throttle output if desired */ diff --git a/src/test-cmp.c b/src/test-cmp.c index e2c5c10b4..3f7c2f47c 100644 --- a/src/test-cmp.c +++ b/src/test-cmp.c @@ -12,6 +12,7 @@ #include #include +#include #include #include #include @@ -128,14 +129,14 @@ check: if (optarg == endptr) serror("Failed to open file: %s", f2.path); while (!feof(f1.handle) && !feof(f2.handle)) { - ret = sample_fscan(f1.handle, f1.sample, &f1.flags); + ret = sample_io_villas_fscan(f1.handle, f1.sample, &f1.flags); if (ret < 0) { if (feof(f1.handle)) ret = 0; goto out; } - ret = sample_fscan(f2.handle, f2.sample, &f2.flags); + ret = sample_io_villas_fscan(f2.handle, f2.sample, &f2.flags); if (ret < 0) { if (feof(f2.handle)) ret = 0; @@ -143,7 +144,7 @@ check: if (optarg == endptr) } /* Compare sequence no */ - if ((f1.flags & SAMPLE_SEQUENCE) && (f2.flags & SAMPLE_SEQUENCE)) { + if ((f1.flags & SAMPLE_IO_SEQUENCE) && (f2.flags & SAMPLE_IO_SEQUENCE)) { if (f1.sample->sequence != f2.sample->sequence) { printf("sequence no: %d != %d\n", f1.sample->sequence, f2.sample->sequence); ret = -1; diff --git a/tests/unit/sample_io.c b/tests/unit/sample_io.c new file mode 100644 index 000000000..9daffb536 --- /dev/null +++ b/tests/unit/sample_io.c @@ -0,0 +1,79 @@ +/** Unit tests for the sample_io module. + * + * @author Steffen Vogel + * @copyright 2017, Institute for Automation of Complex Power Systems, EONERC + *********************************************************************************/ + +#include + +#include +#include + +#include "utils.h" +#include "timing.h" +#include "sample.h" +#include "sample_io.h" + +ParameterizedTestParameters(sample_io, read_write) +{ + static enum sample_io_format formats[] = { + SAMPLE_IO_FORMAT_VILLAS, + SAMPLE_IO_FORMAT_JSON, + }; + + return cr_make_param_array(enum sample_io_format, formats, ARRAY_LEN(formats)); +} + +ParameterizedTest(enum sample_io_format *fmt, sample_io, read_write) +{ + FILE *f; + int ret; + struct sample *s, *t; + + /* Prepare a sample with arbitrary data */ + s = malloc(SAMPLE_LEN(16)); + cr_assert_not_null(s); + + t = malloc(SAMPLE_LEN(16)); + cr_assert_not_null(s); + + t->capacity = + s->capacity = 16; + s->length = 12; + s->sequence = 235; + s->format = 0; + s->ts.origin = time_now(); + s->ts.received = (struct timespec) { s->ts.origin.tv_sec - 1, s->ts.origin.tv_nsec }; + + for (int i = 0; i < s->length; i++) + s->data[i].f = i * 1.2; + + /* Open a file for IO */ + f = tmpfile(); + cr_assert_not_null(f); + + ret = sample_io_fprint(f, s, *fmt, SAMPLE_IO_ALL); + cr_assert_eq(ret, 0); + + rewind(f); + + ret = sample_io_fscan(f, t, *fmt, NULL); + cr_assert_eq(ret, 0); + + cr_assert_eq(s->length, t->length); + cr_assert_eq(s->sequence, t->sequence); + cr_assert_eq(s->format, t->format); + + cr_assert_eq(s->ts.origin.tv_sec, t->ts.origin.tv_sec); + cr_assert_eq(s->ts.origin.tv_nsec, t->ts.origin.tv_nsec); + cr_assert_eq(s->ts.received.tv_sec, t->ts.received.tv_sec); + cr_assert_eq(s->ts.received.tv_nsec, t->ts.received.tv_nsec); + + for (int i = 0; i < MIN(s->length, t->length); i++) + cr_assert_float_eq(s->data[i].f, t->data[i].f, 1e-6); + + fclose(f); + + free(s); + free(t); +} \ No newline at end of file