mirror of
https://git.rwth-aachen.de/acs/public/villas/node/
synced 2025-03-09 00:00:00 +01:00
sample_io: part 2
This commit is contained in:
parent
267d244167
commit
ba5f4d1109
3 changed files with 28 additions and 173 deletions
|
@ -26,13 +26,9 @@ struct pool;
|
|||
/** The offset to the beginning of the data section. */
|
||||
#define SAMPLE_DATA_OFFSET(smp) ((char *) (smp) + offsetof(struct sample, data))
|
||||
|
||||
/** These flags define the format which is used by sample_fscan() and sample_fprint(). */
|
||||
enum sample_flags {
|
||||
SAMPLE_NANOSECONDS = 1,
|
||||
SAMPLE_OFFSET = 2,
|
||||
SAMPLE_SEQUENCE = 4,
|
||||
SAMPLE_VALUES = 8,
|
||||
SAMPLE_ALL = 16-1
|
||||
enum sample_data_format {
|
||||
SAMPLE_DATA_FORMAT_FLOAT= 0,
|
||||
SAMPLE_DATA_FORMAT_INT = 1
|
||||
};
|
||||
|
||||
struct sample {
|
||||
|
@ -50,6 +46,8 @@ struct sample {
|
|||
struct timespec received; /**< The point in time when this data was received. */
|
||||
struct timespec sent; /**< The point in time this data was send for the last time. */
|
||||
} ts;
|
||||
|
||||
uint64_t format; /**< A long bitfield indicating the number representation of the first 64 values in sample::data[] */
|
||||
|
||||
/** The values. */
|
||||
union {
|
||||
|
@ -72,42 +70,8 @@ int sample_get(struct sample *s);
|
|||
/** Decrease reference count and release memory if last reference was held. */
|
||||
int sample_put(struct sample *s);
|
||||
|
||||
/** Print a sample in human readable form to a file stream.
|
||||
*
|
||||
* @param buf A character buffer of len bytes.
|
||||
* @param len The size of buf.
|
||||
* @param m A pointer to the sample.
|
||||
* @param flags See sample_flags.
|
||||
* @return Number of bytes written to buf.
|
||||
*/
|
||||
int sample_print(char *buf, size_t len, struct sample *s, int flags);
|
||||
/** Set number representation for a single value of a sample. */
|
||||
int sample_get_data_format(struct sample *s, int idx);
|
||||
|
||||
/** Read a sample from a character buffer.
|
||||
*
|
||||
* @param line A string which should be parsed.
|
||||
* @param s A pointer to the sample.
|
||||
* @param flags et SAMPLE_* flags for each component found in sample if not NULL. See sample_flags.
|
||||
* @retval 0 Success. Everything went well.
|
||||
* @retval <0 Error. Something went wrong.
|
||||
*/
|
||||
int sample_scan(const char *line, struct sample *s, int *fl);
|
||||
|
||||
/** Print a sample in human readable form to a file stream.
|
||||
*
|
||||
* @see sample_print()
|
||||
* @param f The file handle from fopen() or stdout, stderr.
|
||||
* @param s A pointer to the sample.
|
||||
* @retval 0 Success. Everything went well.
|
||||
* @retval <0 Error. Something went wrong.
|
||||
*/
|
||||
int sample_fprint(FILE *f, struct sample *s, int flags);
|
||||
|
||||
/** Read a sample from a file stream.
|
||||
*
|
||||
* @see sample_scan()
|
||||
* @param f The file handle from fopen() or stdin.
|
||||
* @param s A pointer to the sample.
|
||||
* @retval 0 Success. Everything went well.
|
||||
* @retval <0 Error. Something went wrong.
|
||||
*/
|
||||
int sample_fscan(FILE *f, struct sample *s, int *flags);
|
||||
/** Get number representation for a single value of a sample. */
|
||||
int sample_set_data_format(struct sample *s, int idx, enum sample_data_format fmt);
|
|
@ -12,6 +12,7 @@
|
|||
#include "timing.h"
|
||||
#include "queue.h"
|
||||
#include "plugin.h"
|
||||
#include "sample_io.h"
|
||||
|
||||
int file_reverse(struct node *n)
|
||||
{
|
||||
|
@ -206,7 +207,7 @@ int file_start(struct node *n)
|
|||
|
||||
/* Get timestamp of first line */
|
||||
struct sample s;
|
||||
int ret = sample_fscan(f->read.handle->file, &s, NULL); arewind(f->read.handle);
|
||||
int ret = sample_io_villas_fscan(f->read.handle->file, &s, NULL); arewind(f->read.handle);
|
||||
if (ret < 0)
|
||||
error("Failed to read first timestamp of node %s", node_name(n));
|
||||
|
||||
|
@ -274,7 +275,7 @@ int file_read(struct node *n, struct sample *smps[], unsigned cnt)
|
|||
assert(f->read.handle);
|
||||
assert(cnt == 1);
|
||||
|
||||
retry: values = sample_fscan(f->read.handle->file, s, &flags); /* Get message and timestamp */
|
||||
retry: values = sample_io_villas_fscan(f->read.handle->file, s, &flags); /* Get message and timestamp */
|
||||
if (values < 0) {
|
||||
if (afeof(f->read.handle)) {
|
||||
if (f->read.split) {
|
||||
|
@ -329,7 +330,7 @@ int file_write(struct node *n, struct sample *smps[], unsigned cnt)
|
|||
info("Splitted output node %s: chunk=%u", node_name(n), f->write.chunk);
|
||||
}
|
||||
|
||||
sample_fprint(f->write.handle->file, s, SAMPLE_ALL & ~SAMPLE_OFFSET);
|
||||
sample_io_villas_fprint(f->write.handle->file, s, SAMPLE_IO_ALL & ~SAMPLE_IO_OFFSET);
|
||||
afflush(f->write.handle);
|
||||
|
||||
return 1;
|
||||
|
|
140
lib/sample.c
140
lib/sample.c
|
@ -4,13 +4,11 @@
|
|||
* @copyright 2017, Institute for Automation of Complex Power Systems, EONERC
|
||||
*********************************************************************************/
|
||||
|
||||
#include <ctype.h>
|
||||
|
||||
#include "pool.h"
|
||||
#include "sample.h"
|
||||
#include "timing.h"
|
||||
|
||||
int sample_alloc(struct pool *p, struct sample *smps[], int cnt) {
|
||||
int sample_alloc(struct pool *p, struct sample *smps[], int cnt)
|
||||
{
|
||||
int ret;
|
||||
|
||||
ret = pool_get_many(p, (void **) smps, cnt);
|
||||
|
@ -20,6 +18,7 @@ int sample_alloc(struct pool *p, struct sample *smps[], int cnt) {
|
|||
for (int i = 0; i < ret; i++) {
|
||||
smps[i]->capacity = (p->blocksz - sizeof(**smps)) / sizeof(smps[0]->data[0]);
|
||||
smps[i]->pool = p;
|
||||
smps[i]->format = 0; /* all sample values are float by default */
|
||||
}
|
||||
|
||||
return ret;
|
||||
|
@ -47,132 +46,23 @@ int sample_put(struct sample *s)
|
|||
return prev - 1;
|
||||
}
|
||||
|
||||
int sample_print(char *buf, size_t len, struct sample *s, int flags)
|
||||
int sample_set_data_format(struct sample *s, int idx, enum sample_data_format fmt)
|
||||
{
|
||||
size_t off = snprintf(buf, len, "%llu", (unsigned long long) s->ts.origin.tv_sec);
|
||||
if (idx >= sizeof(s->format) * 8)
|
||||
return 0; /* we currently can only control the format of the first 64 values. */
|
||||
|
||||
if (flags & SAMPLE_NANOSECONDS)
|
||||
off += snprintf(buf + off, len - off, ".%09llu", (unsigned long long) s->ts.origin.tv_nsec);
|
||||
|
||||
if (flags & SAMPLE_OFFSET)
|
||||
off += snprintf(buf + off, len - off, "%+e", time_delta(&s->ts.origin, &s->ts.received));
|
||||
|
||||
if (flags & SAMPLE_SEQUENCE)
|
||||
off += snprintf(buf + off, len - off, "(%u)", s->sequence);
|
||||
|
||||
if (flags & SAMPLE_VALUES) {
|
||||
for (int i = 0; i < s->length; i++)
|
||||
off += snprintf(buf + off, len - off, "\t%.6f", s->data[i].f);
|
||||
switch (fmt) {
|
||||
case SAMPLE_DATA_FORMAT_FLOAT: s->format &= ~(1 << idx); break;
|
||||
case SAMPLE_DATA_FORMAT_INT: s->format |= (fmt << idx); break;
|
||||
}
|
||||
|
||||
off += snprintf(buf + off, len - off, "\n");
|
||||
|
||||
return off + 1; /* trailing '\0' */
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int sample_scan(const char *line, struct sample *s, int *fl)
|
||||
int sample_get_data_format(struct sample *s, int idx)
|
||||
{
|
||||
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_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_OFFSET;
|
||||
else
|
||||
return -4;
|
||||
}
|
||||
if (idx >= sizeof(s->format) * 8)
|
||||
return -1; /* we currently can only control the format of the first 64 values. */
|
||||
|
||||
/* Optional: sequence */
|
||||
if (*end == '(') {
|
||||
ptr = end + 1;
|
||||
|
||||
s->sequence = strtoul(ptr, &end, 10);
|
||||
if (ptr != end)
|
||||
flags |= SAMPLE_SEQUENCE;
|
||||
else
|
||||
return -5;
|
||||
|
||||
if (*end == ')')
|
||||
end++;
|
||||
}
|
||||
|
||||
for (ptr = end, s->length = 0;
|
||||
s->length < s->capacity;
|
||||
ptr = end, s->length++) {
|
||||
|
||||
s->data[s->length].f = strtod(ptr, &end); /** @todo We only support floating point values at the moment */
|
||||
if (end == ptr) /* There are no valid FP values anymore */
|
||||
break;
|
||||
}
|
||||
|
||||
if (s->length > 0)
|
||||
flags |= SAMPLE_VALUES;
|
||||
|
||||
if (fl)
|
||||
*fl = flags;
|
||||
if (flags & SAMPLE_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 s->length;
|
||||
}
|
||||
|
||||
int sample_fprint(FILE *f, struct sample *s, int flags)
|
||||
{
|
||||
char line[4096];
|
||||
|
||||
int len = sample_print(line, sizeof(line), s, flags);
|
||||
|
||||
fputs(line, f);
|
||||
|
||||
return len;
|
||||
}
|
||||
|
||||
int sample_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_scan(line, s, fl);
|
||||
return (s->format >> idx) & 0x1;
|
||||
}
|
Loading…
Add table
Reference in a new issue