1
0
Fork 0
mirror of https://git.rwth-aachen.de/acs/public/villas/node/ synced 2025-03-09 00:00:00 +01:00

added new sample_io module for adding new IO formats

This commit is contained in:
Steffen Vogel 2017-03-27 12:28:13 +02:00
parent 38a983b26e
commit 80d3ae13d5
8 changed files with 458 additions and 10 deletions

View file

@ -0,0 +1,83 @@
/** Read / write sample data in different formats.
*
* @file
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @copyright 2017, Institute for Automation of Complex Power Systems, EONERC
*********************************************************************************/
#pragma once
#include <stdio.h>
#include <jansson.h>
/* 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);

View file

@ -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

282
lib/sample_io.c Normal file
View file

@ -0,0 +1,282 @@
/** The internal datastructure for a sample of simulation data.
*
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @copyright 2017, Institute for Automation of Complex Power Systems, EONERC
*********************************************************************************/
#include <ctype.h>
#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

View file

@ -15,6 +15,7 @@
#include <villas/timing.h>
#include <villas/sample.h>
#include <villas/sample_io.h>
#include <villas/hook.h>
#include <villas/utils.h>
#include <villas/pool.h>
@ -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);

View file

@ -20,6 +20,7 @@
#include <villas/msg.h>
#include <villas/timing.h>
#include <villas/pool.h>
#include <villas/sample_io.h>
#include <villas/kernel/rt.h>
#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();

View file

@ -14,6 +14,7 @@
#include <villas/utils.h>
#include <villas/sample.h>
#include <villas/sample_io.h>
#include <villas/timing.h>
#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 */

View file

@ -12,6 +12,7 @@
#include <jansson.h>
#include <villas/sample.h>
#include <villas/sample_io.h>
#include <villas/utils.h>
#include <villas/hist.h>
#include <villas/timing.h>
@ -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;

79
tests/unit/sample_io.c Normal file
View file

@ -0,0 +1,79 @@
/** Unit tests for the sample_io module.
*
* @author Steffen Vogel <stvogel@eonerc.rwth-aachen.de>
* @copyright 2017, Institute for Automation of Complex Power Systems, EONERC
*********************************************************************************/
#include <stdio.h>
#include <criterion/criterion.h>
#include <criterion/parameterized.h>
#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);
}