mwalbeck/rtl_433
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor file outputs from data

Browse source
This commit is contained in:
Christian W. Zuckschwerdt 2021-12-08 11:11:38 +01:00
parent a26cffa81d
commit 305edc44f9
13 changed files with 838 additions and 784 deletions

4
include/abuf.h
View file

@ -12,8 +12,8 @@
#ifndef INCLUDE_ABUF_H_
#define INCLUDE_ABUF_H_
#if defined _MSC_VER // Microsoft Visual Studio
// MSC has something like C99 restrict as __restrict
#if defined _MSC_VER || defined ESP32 // Microsoft Visual Studio or ESP32
// MSC and ESP32 have something like C99 restrict as __restrict
#ifndef restrict
#define restrict __restrict
#endif

23
include/data.h
View file

@ -21,14 +21,7 @@
#ifndef INCLUDE_DATA_H_
#define INCLUDE_DATA_H_
#include <stdio.h>
#if defined(_MSC_VER) && !defined(__clang__)
// MSVC has something like C99 restrict as __restrict
#ifndef restrict
#define restrict __restrict
#endif
#endif
#include <stddef.h>
typedef enum {
DATA_DATA, /**< pointer to data is stored */
@ -141,20 +134,6 @@ typedef struct data_output {
void (*output_free)(struct data_output *output);
} data_output_t;
/** Construct data output for CSV printer.
@param file the output stream
@return The auxiliary data to pass along with data_csv_printer to data_print.
You must release this object with data_output_free once you're done with it.
*/
struct data_output *data_output_csv_create(FILE *file);
struct data_output *data_output_json_create(FILE *file);
struct data_output *data_output_kv_create(FILE *file);
struct data_output *data_output_syslog_create(const char *host, const char *port);
/** Setup known field keys and start output, used by CSV only.
@param output the data_output handle from data_output_x_create

1
include/decoder.h
View file

@ -6,6 +6,7 @@
#define INCLUDE_DECODER_H_
#include <string.h>
#include <stdio.h>
#include "r_device.h"
#include "bitbuffer.h"
#include "data.h"

6
include/decoder_util.h
View file

@ -17,6 +17,12 @@
#include "data.h"
#include "r_device.h"
#if defined _MSC_VER || defined ESP32 // Microsoft Visual Studio or ESP32
// MSC and ESP32 have something like C99 restrict as __restrict
#ifndef restrict
#define restrict __restrict
#endif
#endif
// Defined in newer <sal.h> for MSVC.
#ifndef _Printf_format_string_
#define _Printf_format_string_

32
include/output_file.h Normal file
View file

@ -0,0 +1,32 @@
/** @file
File outputs for rtl_433 events.
Copyright (C) 2021 Christian Zuckschwerdt
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
*/
#ifndef INCLUDE_OUTPUT_FILE_H_
#define INCLUDE_OUTPUT_FILE_H_
#include "data.h"
#include <stdio.h>
/** Construct data output for CSV printer.
@param file the output stream
@return The auxiliary data to pass along with data_csv_printer to data_print.
You must release this object with data_output_free once you're done with it.
*/
struct data_output *data_output_csv_create(FILE *file);
struct data_output *data_output_json_create(FILE *file);
struct data_output *data_output_kv_create(FILE *file);
struct data_output *data_output_syslog_create(const char *host, const char *port);
#endif /* INCLUDE_OUTPUT_FILE_H_ */

1
src/CMakeLists.txt
View file

@ -22,6 +22,7 @@ add_library(r_433 STATIC
list.c
mongoose.c
optparse.c
output_file.c
output_influx.c
output_mqtt.c
pulse_analyzer.c

767
src/data.c
View file

@ -10,66 +10,21 @@
(at your option) any later version.
*/
#include "data.h"
#include "abuf.h"
#include "fatal.h"
#include <stdarg.h>
#include <assert.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <limits.h>
// gethostname() needs _XOPEN_SOURCE 500 on unistd.h
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE 500
#endif
#ifndef _MSC_VER
#include <unistd.h>
#endif
#ifdef _WIN32
#if !defined(_WIN32_WINNT) || (_WIN32_WINNT < 0x0600)
#undef _WIN32_WINNT
#define _WIN32_WINNT 0x0600 /* Needed to pull in 'struct sockaddr_storage' */
#endif
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#define SOCKET int
#define INVALID_SOCKET (-1)
#define closesocket(x) close(x)
#endif
#include <time.h>
#include "term_ctl.h"
#include "abuf.h"
#include "fatal.h"
#include "r_util.h"
#include "data.h"
#ifdef _WIN32
#define _POSIX_HOST_NAME_MAX 128
#define perror(str) ws2_perror(str)
static void ws2_perror (const char *str)
{
if (str && *str)
fprintf(stderr, "%s: ", str);
fprintf(stderr, "Winsock error %d.\n", WSAGetLastError());
}
#endif
#ifdef ESP32
#include <tcpip_adapter.h>
#define _POSIX_HOST_NAME_MAX 128
#define gai_strerror strerror
#endif
// Macro to prevent unused variables (passed into a function)
// from generating a warning.
#define UNUSED(x) (void)(x)
typedef void* (*array_elementwise_import_fn)(void*);
typedef void (*array_element_release_fn)(void*);
@ -456,548 +411,6 @@ void print_array_value(data_output_t *output, data_array_t *array, char const *f
}
}
/* JSON printer */
typedef struct {
struct data_output output;
FILE *file;
} data_output_json_t;
static void print_json_array(data_output_t *output, data_array_t *array, char const *format)
{
data_output_json_t *json = (data_output_json_t *)output;
fprintf(json->file, "[");
for (int c = 0; c < array->num_values; ++c) {
if (c)
fprintf(json->file, ", ");
print_array_value(output, array, format, c);
}
fprintf(json->file, "]");
}
static void print_json_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
bool separator = false;
fputc('{', json->file);
while (data) {
if (separator)
fprintf(json->file, ", ");
output->print_string(output, data->key, NULL);
fprintf(json->file, " : ");
print_value(output, data->type, data->value, data->format);
separator = true;
data = data->next;
}
fputc('}', json->file);
}
static void print_json_string(data_output_t *output, const char *str, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
size_t str_len = strlen(str);
if (str[0] == '{' && str[str_len - 1] == '}') {
// Print embedded JSON object verbatim
fprintf(json->file, "%s", str);
return;
}
fprintf(json->file, "\"");
while (*str) {
if (*str == '\r') {
fprintf(json->file, "\\r");
continue;
}
if (*str == '\n') {
fprintf(json->file, "\\n");
continue;
}
if (*str == '\t') {
fprintf(json->file, "\\t");
continue;
}
if (*str == '"' || *str == '\\')
fputc('\\', json->file);
fputc(*str, json->file);
++str;
}
fprintf(json->file, "\"");
}
static void print_json_double(data_output_t *output, double data, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
fprintf(json->file, "%.3f", data);
}
static void print_json_int(data_output_t *output, int data, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
fprintf(json->file, "%d", data);
}
static void print_json_flush(data_output_t *output)
{
data_output_json_t *json = (data_output_json_t *)output;
if (json && json->file) {
fputc('\n', json->file);
fflush(json->file);
}
}
static void data_output_json_free(data_output_t *output)
{
if (!output)
return;
free(output);
}
struct data_output *data_output_json_create(FILE *file)
{
data_output_json_t *json = calloc(1, sizeof(data_output_json_t));
if (!json) {
WARN_CALLOC("data_output_json_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
json->output.print_data = print_json_data;
json->output.print_array = print_json_array;
json->output.print_string = print_json_string;
json->output.print_double = print_json_double;
json->output.print_int = print_json_int;
json->output.output_flush = print_json_flush;
json->output.output_free = data_output_json_free;
json->file = file;
return &json->output;
}
/* Pretty Key-Value printer */
static int kv_color_for_key(char const *key)
{
if (!key || !*key)
return TERM_COLOR_RESET;
if (!strcmp(key, "tag") || !strcmp(key, "time"))
return TERM_COLOR_BLUE;
if (!strcmp(key, "model") || !strcmp(key, "type") || !strcmp(key, "id"))
return TERM_COLOR_RED;
if (!strcmp(key, "mic"))
return TERM_COLOR_CYAN;
if (!strcmp(key, "mod") || !strcmp(key, "freq") || !strcmp(key, "freq1") || !strcmp(key, "freq2"))
return TERM_COLOR_MAGENTA;
if (!strcmp(key, "rssi") || !strcmp(key, "snr") || !strcmp(key, "noise"))
return TERM_COLOR_YELLOW;
return TERM_COLOR_GREEN;
}
static int kv_break_before_key(char const *key)
{
if (!key || !*key)
return 0;
if (!strcmp(key, "model") || !strcmp(key, "mod") || !strcmp(key, "rssi") || !strcmp(key, "codes"))
return 1;
return 0;
}
static int kv_break_after_key(char const *key)
{
if (!key || !*key)
return 0;
if (!strcmp(key, "id") || !strcmp(key, "mic"))
return 1;
return 0;
}
typedef struct {
struct data_output output;
FILE *file;
void *term;
int color;
int ring_bell;
int term_width;
int data_recursion;
int column;
} data_output_kv_t;
#define KV_SEP "_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ "
static void print_kv_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_kv_t *kv = (data_output_kv_t *)output;
int color = kv->color;
int ring_bell = kv->ring_bell;
// top-level: update width and print separator
if (!kv->data_recursion) {
kv->term_width = term_get_columns(kv->term); // update current term width
if (color)
term_set_fg(kv->term, TERM_COLOR_BLACK);
if (ring_bell)
term_ring_bell(kv->term);
char sep[] = KV_SEP KV_SEP KV_SEP KV_SEP;
if (kv->term_width < (int)sizeof(sep))
sep[kv->term_width > 0 ? kv->term_width - 1 : 40] = '\0';
fprintf(kv->file, "%s\n", sep);
if (color)
term_set_fg(kv->term, TERM_COLOR_RESET);
}
// nested data object: break before
else {
if (color)
term_set_fg(kv->term, TERM_COLOR_RESET);
fprintf(kv->file, "\n");
kv->column = 0;
}
++kv->data_recursion;
while (data) {
// break before some known keys
if (kv->column > 0 && kv_break_before_key(data->key)) {
fprintf(kv->file, "\n");
kv->column = 0;
}
// break if not enough width left
else if (kv->column >= kv->term_width - 26) {
fprintf(kv->file, "\n");
kv->column = 0;
}
// pad to next alignment if there is enough width left
else if (kv->column > 0 && kv->column < kv->term_width - 26) {
kv->column += fprintf(kv->file, "%*s", 25 - kv->column % 26, " ");
}
// print key
char *key = *data->pretty_key ? data->pretty_key : data->key;
kv->column += fprintf(kv->file, "%-10s: ", key);
// print value
if (color)
term_set_fg(kv->term, kv_color_for_key(data->key));
print_value(output, data->type, data->value, data->format);
if (color)
term_set_fg(kv->term, TERM_COLOR_RESET);
// force break after some known keys
if (kv->column > 0 && kv_break_after_key(data->key)) {
kv->column = kv->term_width; // force break;
}
data = data->next;
}
--kv->data_recursion;
// top-level: always end with newline
if (!kv->data_recursion && kv->column > 0) {
//fprintf(kv->file, "\n"); // data_output_print() already adds a newline
kv->column = 0;
}
}
static void print_kv_array(data_output_t *output, data_array_t *array, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
//fprintf(kv->file, "[ ");
for (int c = 0; c < array->num_values; ++c) {
if (c)
fprintf(kv->file, ", ");
print_array_value(output, array, format, c);
}
//fprintf(kv->file, " ]");
}
static void print_kv_double(data_output_t *output, double data, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
kv->column += fprintf(kv->file, format ? format : "%.3f", data);
}
static void print_kv_int(data_output_t *output, int data, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
kv->column += fprintf(kv->file, format ? format : "%d", data);
}
static void print_kv_string(data_output_t *output, const char *data, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
kv->column += fprintf(kv->file, format ? format : "%s", data);
}
static void print_kv_flush(data_output_t *output)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
if (kv && kv->file) {
fputc('\n', kv->file);
fflush(kv->file);
}
}
static void data_output_kv_free(data_output_t *output)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
if (!output)
return;
if (kv->color)
term_free(kv->term);
free(output);
}
struct data_output *data_output_kv_create(FILE *file)
{
data_output_kv_t *kv = calloc(1, sizeof(data_output_kv_t));
if (!kv) {
WARN_CALLOC("data_output_kv_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
kv->output.print_data = print_kv_data;
kv->output.print_array = print_kv_array;
kv->output.print_string = print_kv_string;
kv->output.print_double = print_kv_double;
kv->output.print_int = print_kv_int;
kv->output.output_flush = print_kv_flush;
kv->output.output_free = data_output_kv_free;
kv->file = file;
kv->term = term_init(file);
kv->color = term_has_color(kv->term);
kv->ring_bell = 0; // TODO: enable if requested...
return &kv->output;
}
/* CSV printer; doesn't really support recursive data objects yet */
typedef struct {
struct data_output output;
FILE *file;
const char **fields;
int data_recursion;
const char *separator;
} data_output_csv_t;
static void print_csv_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
const char **fields = csv->fields;
int i;
if (csv->data_recursion)
return;
int regular = 0; // skip "states" output
for (data_t *d = data; d; d = d->next) {
if (!strcmp(d->key, "msg") || !strcmp(d->key, "codes") || !strcmp(d->key, "model")) {
regular = 1;
break;
}
}
if (!regular)
return;
++csv->data_recursion;
for (i = 0; fields[i]; ++i) {
const char *key = fields[i];
data_t *found = NULL;
if (i)
fprintf(csv->file, "%s", csv->separator);
for (data_t *iter = data; !found && iter; iter = iter->next)
if (strcmp(iter->key, key) == 0)
found = iter;
if (found)
print_value(output, found->type, found->value, found->format);
}
--csv->data_recursion;
}
static void print_csv_array(data_output_t *output, data_array_t *array, char const *format)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
for (int c = 0; c < array->num_values; ++c) {
if (c)
fprintf(csv->file, ";");
print_array_value(output, array, format, c);
}
}
static void print_csv_string(data_output_t *output, const char *str, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
while (*str) {
if (strncmp(str, csv->separator, strlen(csv->separator)) == 0)
fputc('\\', csv->file);
fputc(*str, csv->file);
++str;
}
}
static int compare_strings(const void *a, const void *b)
{
return strcmp(*(char **)a, *(char **)b);
}
static void data_output_csv_start(struct data_output *output, char const *const *fields, int num_fields)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
int csv_fields = 0;
int i, j;
const char **allowed = NULL;
int *use_count = NULL;
int num_unique_fields;
if (!csv)
goto alloc_error;
csv->separator = ",";
allowed = calloc(num_fields, sizeof(const char *));
if (!allowed) {
WARN_CALLOC("data_output_csv_start()");
goto alloc_error;
}
memcpy((void *)allowed, fields, sizeof(const char *) * num_fields);
qsort((void *)allowed, num_fields, sizeof(char *), compare_strings);
// overwrite duplicates
i = 0;
j = 0;
while (j < num_fields) {
while (j > 0 && j < num_fields &&
strcmp(allowed[j - 1], allowed[j]) == 0)
++j;
if (j < num_fields) {
allowed[i] = allowed[j];
++i;
++j;
}
}
num_unique_fields = i;
csv->fields = calloc(num_unique_fields + 1, sizeof(const char *));
if (!csv->fields) {
WARN_CALLOC("data_output_csv_start()");
goto alloc_error;
}
use_count = calloc(num_unique_fields + 1, sizeof(*use_count)); // '+ 1' so we never alloc size 0
if (!use_count) {
WARN_CALLOC("data_output_csv_start()");
goto alloc_error;
}
for (i = 0; i < num_fields; ++i) {
const char **field = bsearch(&fields[i], allowed, num_unique_fields, sizeof(const char *),
compare_strings);
int *field_use_count = use_count + (field - allowed);
if (field && !*field_use_count) {
csv->fields[csv_fields] = fields[i];
++csv_fields;
++*field_use_count;
}
}
csv->fields[csv_fields] = NULL;
free((void *)allowed);
free(use_count);
// Output the CSV header
for (i = 0; csv->fields[i]; ++i) {
fprintf(csv->file, "%s%s", i > 0 ? csv->separator : "", csv->fields[i]);
}
fprintf(csv->file, "\n");
return;
alloc_error:
free(use_count);
free((void *)allowed);
if (csv)
free((void *)csv->fields);
free(csv);
}
static void print_csv_double(data_output_t *output, double data, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
fprintf(csv->file, "%.3f", data);
}
static void print_csv_int(data_output_t *output, int data, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
fprintf(csv->file, "%d", data);
}
static void print_csv_flush(data_output_t *output)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
if (csv && csv->file) {
fputc('\n', csv->file);
fflush(csv->file);
}
}
static void data_output_csv_free(data_output_t *output)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
free((void *)csv->fields);
free(csv);
}
struct data_output *data_output_csv_create(FILE *file)
{
data_output_csv_t *csv = calloc(1, sizeof(data_output_csv_t));
if (!csv) {
WARN_CALLOC("data_output_csv_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
csv->output.print_data = print_csv_data;
csv->output.print_array = print_csv_array;
csv->output.print_string = print_csv_string;
csv->output.print_double = print_csv_double;
csv->output.print_int = print_csv_int;
csv->output.output_start = data_output_csv_start;
csv->output.output_flush = print_csv_flush;
csv->output.output_free = data_output_csv_free;
csv->file = file;
return &csv->output;
}
/* JSON string printer */
typedef struct {
@ -1141,167 +554,3 @@ size_t data_print_jsons(data_t *data, char *dst, size_t len)
return len - jsons.msg.left;
}
/* Datagram (UDP) client */
typedef struct {
struct sockaddr_storage addr;
socklen_t addr_len;
SOCKET sock;
} datagram_client_t;
static int datagram_client_open(datagram_client_t *client, const char *host, const char *port)
{
if (!host || !port)
return -1;
struct addrinfo hints, *res, *res0;
int error;
SOCKET sock;
memset(&hints, 0, sizeof(hints));
hints.ai_family = PF_UNSPEC;
hints.ai_socktype = SOCK_DGRAM;
hints.ai_flags = AI_ADDRCONFIG;
error = getaddrinfo(host, port, &hints, &res0);
if (error) {
fprintf(stderr, "%s\n", gai_strerror(error));
return -1;
}
sock = INVALID_SOCKET;
for (res = res0; res; res = res->ai_next) {
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock >= 0) {
client->sock = sock;
memset(&client->addr, 0, sizeof(client->addr));
memcpy(&client->addr, res->ai_addr, res->ai_addrlen);
client->addr_len = res->ai_addrlen;
break; // success
}
}
freeaddrinfo(res0);
if (sock == INVALID_SOCKET) {
perror("socket");
return -1;
}
//int broadcast = 1;
//int ret = setsockopt(client->sock, SOL_SOCKET, SO_BROADCAST, &broadcast, sizeof(broadcast));
return 0;
}
static void datagram_client_close(datagram_client_t *client)
{
if (!client)
return;
if (client->sock != INVALID_SOCKET) {
closesocket(client->sock);
client->sock = INVALID_SOCKET;
}
#ifdef _WIN32
WSACleanup();
#endif
}
static void datagram_client_send(datagram_client_t *client, const char *message, size_t message_len)
{
int r = sendto(client->sock, message, message_len, 0, (struct sockaddr *)&client->addr, client->addr_len);
if (r == -1) {
perror("sendto");
}
}
/* Syslog UDP printer, RFC 5424 (IETF-syslog protocol) */
typedef struct {
struct data_output output;
datagram_client_t client;
int pri;
char hostname[_POSIX_HOST_NAME_MAX + 1];
} data_output_syslog_t;
static void print_syslog_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_syslog_t *syslog = (data_output_syslog_t *)output;
// we expect a normal message around 500 bytes
// full stats report would be 12k and we want a max of MTU anyway
char message[1024];
abuf_t msg = {0};
abuf_init(&msg, message, sizeof(message));
time_t now;
struct tm tm_info;
time(&now);
#ifdef _WIN32
gmtime_s(&tm_info, &now);
#else
gmtime_r(&now, &tm_info);
#endif
char timestamp[21];
strftime(timestamp, 21, "%Y-%m-%dT%H:%M:%SZ", &tm_info);
abuf_printf(&msg, "<%d>1 %s %s rtl_433 - - - ", syslog->pri, timestamp, syslog->hostname);
msg.tail += data_print_jsons(data, msg.tail, msg.left);
if (msg.tail >= msg.head + sizeof(message))
return; // abort on overflow, we don't actually want to send more than fits the MTU
size_t abuf_len = msg.tail - msg.head;
datagram_client_send(&syslog->client, message, abuf_len);
}
static void data_output_syslog_free(data_output_t *output)
{
data_output_syslog_t *syslog = (data_output_syslog_t *)output;
if (!syslog)
return;
datagram_client_close(&syslog->client);
free(syslog);
}
struct data_output *data_output_syslog_create(const char *host, const char *port)
{
data_output_syslog_t *syslog = calloc(1, sizeof(data_output_syslog_t));
if (!syslog) {
WARN_CALLOC("data_output_syslog_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
#ifdef _WIN32
WSADATA wsa;
if (WSAStartup(MAKEWORD(2,2),&wsa) != 0) {
perror("WSAStartup()");
free(syslog);
return NULL;
}
#endif
syslog->output.print_data = print_syslog_data;
syslog->output.output_free = data_output_syslog_free;
// Severity 5 "Notice", Facility 20 "local use 4"
syslog->pri = 20 * 8 + 5;
#ifdef ESP32
const char* adapter_hostname = NULL;
tcpip_adapter_get_hostname(TCPIP_ADAPTER_IF_STA, &adapter_hostname);
if (adapter_hostname) {
memcpy(syslog->hostname, adapter_hostname, _POSIX_HOST_NAME_MAX);
}
else {
syslog->hostname[0] = '\0';
}
#else
gethostname(syslog->hostname, _POSIX_HOST_NAME_MAX + 1);
#endif
syslog->hostname[_POSIX_HOST_NAME_MAX] = '\0';
datagram_client_open(&syslog->client, host, port);
return &syslog->output;
}

776
src/output_file.c Normal file
View file

@ -0,0 +1,776 @@
/** @file
File outputs for rtl_433 events.
Copyright (C) 2021 Christian Zuckschwerdt
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
*/
#include "output_file.h"
#include "data.h"
#include "term_ctl.h"
#include "abuf.h"
#include "r_util.h"
#include "fatal.h"
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <limits.h>
// gethostname() needs _XOPEN_SOURCE 500 on unistd.h
#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE 500
#endif
#ifndef _MSC_VER
#include <unistd.h>
#endif
#ifdef _WIN32
#if !defined(_WIN32_WINNT) || (_WIN32_WINNT < 0x0600)
#undef _WIN32_WINNT
#define _WIN32_WINNT 0x0600 /* Needed to pull in 'struct sockaddr_storage' */
#endif
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <netinet/in.h>
#define SOCKET int
#define INVALID_SOCKET (-1)
#define closesocket(x) close(x)
#endif
#include <time.h>
#ifdef _WIN32
#define _POSIX_HOST_NAME_MAX 128
#define perror(str) ws2_perror(str)
static void ws2_perror (const char *str)
{
if (str && *str)
fprintf(stderr, "%s: ", str);
fprintf(stderr, "Winsock error %d.\n", WSAGetLastError());
}
#endif
#ifdef ESP32
#include <tcpip_adapter.h>
#define _POSIX_HOST_NAME_MAX 128
#define gai_strerror strerror
#endif
/* JSON printer */
typedef struct {
struct data_output output;
FILE *file;
} data_output_json_t;
static void print_json_array(data_output_t *output, data_array_t *array, char const *format)
{
data_output_json_t *json = (data_output_json_t *)output;
fprintf(json->file, "[");
for (int c = 0; c < array->num_values; ++c) {
if (c)
fprintf(json->file, ", ");
print_array_value(output, array, format, c);
}
fprintf(json->file, "]");
}
static void print_json_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
bool separator = false;
fputc('{', json->file);
while (data) {
if (separator)
fprintf(json->file, ", ");
output->print_string(output, data->key, NULL);
fprintf(json->file, " : ");
print_value(output, data->type, data->value, data->format);
separator = true;
data = data->next;
}
fputc('}', json->file);
}
static void print_json_string(data_output_t *output, const char *str, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
size_t str_len = strlen(str);
if (str[0] == '{' && str[str_len - 1] == '}') {
// Print embedded JSON object verbatim
fprintf(json->file, "%s", str);
return;
}
fprintf(json->file, "\"");
while (*str) {
if (*str == '\r') {
fprintf(json->file, "\\r");
continue;
}
if (*str == '\n') {
fprintf(json->file, "\\n");
continue;
}
if (*str == '\t') {
fprintf(json->file, "\\t");
continue;
}
if (*str == '"' || *str == '\\')
fputc('\\', json->file);
fputc(*str, json->file);
++str;
}
fprintf(json->file, "\"");
}
static void print_json_double(data_output_t *output, double data, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
fprintf(json->file, "%.3f", data);
}
static void print_json_int(data_output_t *output, int data, char const *format)
{
UNUSED(format);
data_output_json_t *json = (data_output_json_t *)output;
fprintf(json->file, "%d", data);
}
static void print_json_flush(data_output_t *output)
{
data_output_json_t *json = (data_output_json_t *)output;
if (json && json->file) {
fputc('\n', json->file);
fflush(json->file);
}
}
static void data_output_json_free(data_output_t *output)
{
if (!output)
return;
free(output);
}
struct data_output *data_output_json_create(FILE *file)
{
data_output_json_t *json = calloc(1, sizeof(data_output_json_t));
if (!json) {
WARN_CALLOC("data_output_json_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
json->output.print_data = print_json_data;
json->output.print_array = print_json_array;
json->output.print_string = print_json_string;
json->output.print_double = print_json_double;
json->output.print_int = print_json_int;
json->output.output_flush = print_json_flush;
json->output.output_free = data_output_json_free;
json->file = file;
return &json->output;
}
/* Pretty Key-Value printer */
static int kv_color_for_key(char const *key)
{
if (!key || !*key)
return TERM_COLOR_RESET;
if (!strcmp(key, "tag") || !strcmp(key, "time"))
return TERM_COLOR_BLUE;
if (!strcmp(key, "model") || !strcmp(key, "type") || !strcmp(key, "id"))
return TERM_COLOR_RED;
if (!strcmp(key, "mic"))
return TERM_COLOR_CYAN;
if (!strcmp(key, "mod") || !strcmp(key, "freq") || !strcmp(key, "freq1") || !strcmp(key, "freq2"))
return TERM_COLOR_MAGENTA;
if (!strcmp(key, "rssi") || !strcmp(key, "snr") || !strcmp(key, "noise"))
return TERM_COLOR_YELLOW;
return TERM_COLOR_GREEN;
}
static int kv_break_before_key(char const *key)
{
if (!key || !*key)
return 0;
if (!strcmp(key, "model") || !strcmp(key, "mod") || !strcmp(key, "rssi") || !strcmp(key, "codes"))
return 1;
return 0;
}
static int kv_break_after_key(char const *key)
{
if (!key || !*key)
return 0;
if (!strcmp(key, "id") || !strcmp(key, "mic"))
return 1;
return 0;
}
typedef struct {
struct data_output output;
FILE *file;
void *term;
int color;
int ring_bell;
int term_width;
int data_recursion;
int column;
} data_output_kv_t;
#define KV_SEP "_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ "
static void print_kv_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_kv_t *kv = (data_output_kv_t *)output;
int color = kv->color;
int ring_bell = kv->ring_bell;
// top-level: update width and print separator
if (!kv->data_recursion) {
kv->term_width = term_get_columns(kv->term); // update current term width
if (color)
term_set_fg(kv->term, TERM_COLOR_BLACK);
if (ring_bell)
term_ring_bell(kv->term);
char sep[] = KV_SEP KV_SEP KV_SEP KV_SEP;
if (kv->term_width < (int)sizeof(sep))
sep[kv->term_width > 0 ? kv->term_width - 1 : 40] = '\0';
fprintf(kv->file, "%s\n", sep);
if (color)
term_set_fg(kv->term, TERM_COLOR_RESET);
}
// nested data object: break before
else {
if (color)
term_set_fg(kv->term, TERM_COLOR_RESET);
fprintf(kv->file, "\n");
kv->column = 0;
}
++kv->data_recursion;
while (data) {
// break before some known keys
if (kv->column > 0 && kv_break_before_key(data->key)) {
fprintf(kv->file, "\n");
kv->column = 0;
}
// break if not enough width left
else if (kv->column >= kv->term_width - 26) {
fprintf(kv->file, "\n");
kv->column = 0;
}
// pad to next alignment if there is enough width left
else if (kv->column > 0 && kv->column < kv->term_width - 26) {
kv->column += fprintf(kv->file, "%*s", 25 - kv->column % 26, " ");
}
// print key
char *key = *data->pretty_key ? data->pretty_key : data->key;
kv->column += fprintf(kv->file, "%-10s: ", key);
// print value
if (color)
term_set_fg(kv->term, kv_color_for_key(data->key));
print_value(output, data->type, data->value, data->format);
if (color)
term_set_fg(kv->term, TERM_COLOR_RESET);
// force break after some known keys
if (kv->column > 0 && kv_break_after_key(data->key)) {
kv->column = kv->term_width; // force break;
}
data = data->next;
}
--kv->data_recursion;
// top-level: always end with newline
if (!kv->data_recursion && kv->column > 0) {
//fprintf(kv->file, "\n"); // data_output_print() already adds a newline
kv->column = 0;
}
}
static void print_kv_array(data_output_t *output, data_array_t *array, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
//fprintf(kv->file, "[ ");
for (int c = 0; c < array->num_values; ++c) {
if (c)
fprintf(kv->file, ", ");
print_array_value(output, array, format, c);
}
//fprintf(kv->file, " ]");
}
static void print_kv_double(data_output_t *output, double data, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
kv->column += fprintf(kv->file, format ? format : "%.3f", data);
}
static void print_kv_int(data_output_t *output, int data, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
kv->column += fprintf(kv->file, format ? format : "%d", data);
}
static void print_kv_string(data_output_t *output, const char *data, char const *format)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
kv->column += fprintf(kv->file, format ? format : "%s", data);
}
static void print_kv_flush(data_output_t *output)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
if (kv && kv->file) {
fputc('\n', kv->file);
fflush(kv->file);
}
}
static void data_output_kv_free(data_output_t *output)
{
data_output_kv_t *kv = (data_output_kv_t *)output;
if (!output)
return;
if (kv->color)
term_free(kv->term);
free(output);
}
struct data_output *data_output_kv_create(FILE *file)
{
data_output_kv_t *kv = calloc(1, sizeof(data_output_kv_t));
if (!kv) {
WARN_CALLOC("data_output_kv_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
kv->output.print_data = print_kv_data;
kv->output.print_array = print_kv_array;
kv->output.print_string = print_kv_string;
kv->output.print_double = print_kv_double;
kv->output.print_int = print_kv_int;
kv->output.output_flush = print_kv_flush;
kv->output.output_free = data_output_kv_free;
kv->file = file;
kv->term = term_init(file);
kv->color = term_has_color(kv->term);
kv->ring_bell = 0; // TODO: enable if requested...
return &kv->output;
}
/* CSV printer; doesn't really support recursive data objects yet */
typedef struct {
struct data_output output;
FILE *file;
const char **fields;
int data_recursion;
const char *separator;
} data_output_csv_t;
static void print_csv_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
const char **fields = csv->fields;
int i;
if (csv->data_recursion)
return;
int regular = 0; // skip "states" output
for (data_t *d = data; d; d = d->next) {
if (!strcmp(d->key, "msg") || !strcmp(d->key, "codes") || !strcmp(d->key, "model")) {
regular = 1;
break;
}
}
if (!regular)
return;
++csv->data_recursion;
for (i = 0; fields[i]; ++i) {
const char *key = fields[i];
data_t *found = NULL;
if (i)
fprintf(csv->file, "%s", csv->separator);
for (data_t *iter = data; !found && iter; iter = iter->next)
if (strcmp(iter->key, key) == 0)
found = iter;
if (found)
print_value(output, found->type, found->value, found->format);
}
--csv->data_recursion;
}
static void print_csv_array(data_output_t *output, data_array_t *array, char const *format)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
for (int c = 0; c < array->num_values; ++c) {
if (c)
fprintf(csv->file, ";");
print_array_value(output, array, format, c);
}
}
static void print_csv_string(data_output_t *output, const char *str, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
while (*str) {
if (strncmp(str, csv->separator, strlen(csv->separator)) == 0)
fputc('\\', csv->file);
fputc(*str, csv->file);
++str;
}
}
static int compare_strings(const void *a, const void *b)
{
return strcmp(*(char **)a, *(char **)b);
}
static void data_output_csv_start(struct data_output *output, char const *const *fields, int num_fields)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
int csv_fields = 0;
int i, j;
const char **allowed = NULL;
int *use_count = NULL;
int num_unique_fields;
if (!csv)
goto alloc_error;
csv->separator = ",";
allowed = calloc(num_fields, sizeof(const char *));
if (!allowed) {
WARN_CALLOC("data_output_csv_start()");
goto alloc_error;
}
memcpy((void *)allowed, fields, sizeof(const char *) * num_fields);
qsort((void *)allowed, num_fields, sizeof(char *), compare_strings);
// overwrite duplicates
i = 0;
j = 0;
while (j < num_fields) {
while (j > 0 && j < num_fields &&
strcmp(allowed[j - 1], allowed[j]) == 0)
++j;
if (j < num_fields) {
allowed[i] = allowed[j];
++i;
++j;
}
}
num_unique_fields = i;
csv->fields = calloc(num_unique_fields + 1, sizeof(const char *));
if (!csv->fields) {
WARN_CALLOC("data_output_csv_start()");
goto alloc_error;
}
use_count = calloc(num_unique_fields + 1, sizeof(*use_count)); // '+ 1' so we never alloc size 0
if (!use_count) {
WARN_CALLOC("data_output_csv_start()");
goto alloc_error;
}
for (i = 0; i < num_fields; ++i) {
const char **field = bsearch(&fields[i], allowed, num_unique_fields, sizeof(const char *),
compare_strings);
int *field_use_count = use_count + (field - allowed);
if (field && !*field_use_count) {
csv->fields[csv_fields] = fields[i];
++csv_fields;
++*field_use_count;
}
}
csv->fields[csv_fields] = NULL;
free((void *)allowed);
free(use_count);
// Output the CSV header
for (i = 0; csv->fields[i]; ++i) {
fprintf(csv->file, "%s%s", i > 0 ? csv->separator : "", csv->fields[i]);
}
fprintf(csv->file, "\n");
return;
alloc_error:
free(use_count);
free((void *)allowed);
if (csv)
free((void *)csv->fields);
free(csv);
}
static void print_csv_double(data_output_t *output, double data, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
fprintf(csv->file, "%.3f", data);
}
static void print_csv_int(data_output_t *output, int data, char const *format)
{
UNUSED(format);
data_output_csv_t *csv = (data_output_csv_t *)output;
fprintf(csv->file, "%d", data);
}
static void print_csv_flush(data_output_t *output)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
if (csv && csv->file) {
fputc('\n', csv->file);
fflush(csv->file);
}
}
static void data_output_csv_free(data_output_t *output)
{
data_output_csv_t *csv = (data_output_csv_t *)output;
free((void *)csv->fields);
free(csv);
}
struct data_output *data_output_csv_create(FILE *file)
{
data_output_csv_t *csv = calloc(1, sizeof(data_output_csv_t));
if (!csv) {
WARN_CALLOC("data_output_csv_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
csv->output.print_data = print_csv_data;
csv->output.print_array = print_csv_array;
csv->output.print_string = print_csv_string;
csv->output.print_double = print_csv_double;
csv->output.print_int = print_csv_int;
csv->output.output_start = data_output_csv_start;
csv->output.output_flush = print_csv_flush;
csv->output.output_free = data_output_csv_free;
csv->file = file;
return &csv->output;
}
/* Datagram (UDP) client */
typedef struct {
struct sockaddr_storage addr;
socklen_t addr_len;
SOCKET sock;
} datagram_client_t;
static int datagram_client_open(datagram_client_t *client, const char *host, const char *port)
{
if (!host || !port)
return -1;
struct addrinfo hints, *res, *res0;
int error;
SOCKET sock;
memset(&hints, 0, sizeof(hints));
hints.ai_family = PF_UNSPEC;
hints.ai_socktype = SOCK_DGRAM;
hints.ai_flags = AI_ADDRCONFIG;
error = getaddrinfo(host, port, &hints, &res0);
if (error) {
fprintf(stderr, "%s\n", gai_strerror(error));
return -1;
}
sock = INVALID_SOCKET;
for (res = res0; res; res = res->ai_next) {
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock >= 0) {
client->sock = sock;
memset(&client->addr, 0, sizeof(client->addr));
memcpy(&client->addr, res->ai_addr, res->ai_addrlen);
client->addr_len = res->ai_addrlen;
break; // success
}
}
freeaddrinfo(res0);
if (sock == INVALID_SOCKET) {
perror("socket");
return -1;
}
//int broadcast = 1;
//int ret = setsockopt(client->sock, SOL_SOCKET, SO_BROADCAST, &broadcast, sizeof(broadcast));
return 0;
}
static void datagram_client_close(datagram_client_t *client)
{
if (!client)
return;
if (client->sock != INVALID_SOCKET) {
closesocket(client->sock);
client->sock = INVALID_SOCKET;
}
#ifdef _WIN32
WSACleanup();
#endif
}
static void datagram_client_send(datagram_client_t *client, const char *message, size_t message_len)
{
int r = sendto(client->sock, message, message_len, 0, (struct sockaddr *)&client->addr, client->addr_len);
if (r == -1) {
perror("sendto");
}
}
/* Syslog UDP printer, RFC 5424 (IETF-syslog protocol) */
typedef struct {
struct data_output output;
datagram_client_t client;
int pri;
char hostname[_POSIX_HOST_NAME_MAX + 1];
} data_output_syslog_t;
static void print_syslog_data(data_output_t *output, data_t *data, char const *format)
{
UNUSED(format);
data_output_syslog_t *syslog = (data_output_syslog_t *)output;
// we expect a normal message around 500 bytes
// full stats report would be 12k and we want a max of MTU anyway
char message[1024];
abuf_t msg = {0};
abuf_init(&msg, message, sizeof(message));
time_t now;
struct tm tm_info;
time(&now);
#ifdef _WIN32
gmtime_s(&tm_info, &now);
#else
gmtime_r(&now, &tm_info);
#endif
char timestamp[21];
strftime(timestamp, 21, "%Y-%m-%dT%H:%M:%SZ", &tm_info);
abuf_printf(&msg, "<%d>1 %s %s rtl_433 - - - ", syslog->pri, timestamp, syslog->hostname);
msg.tail += data_print_jsons(data, msg.tail, msg.left);
if (msg.tail >= msg.head + sizeof(message))
return; // abort on overflow, we don't actually want to send more than fits the MTU
size_t abuf_len = msg.tail - msg.head;
datagram_client_send(&syslog->client, message, abuf_len);
}
static void data_output_syslog_free(data_output_t *output)
{
data_output_syslog_t *syslog = (data_output_syslog_t *)output;
if (!syslog)
return;
datagram_client_close(&syslog->client);
free(syslog);
}
struct data_output *data_output_syslog_create(const char *host, const char *port)
{
data_output_syslog_t *syslog = calloc(1, sizeof(data_output_syslog_t));
if (!syslog) {
WARN_CALLOC("data_output_syslog_create()");
return NULL; // NOTE: returns NULL on alloc failure.
}
#ifdef _WIN32
WSADATA wsa;
if (WSAStartup(MAKEWORD(2,2),&wsa) != 0) {
perror("WSAStartup()");
free(syslog);
return NULL;
}
#endif
syslog->output.print_data = print_syslog_data;
syslog->output.output_free = data_output_syslog_free;
// Severity 5 "Notice", Facility 20 "local use 4"
syslog->pri = 20 * 8 + 5;
#ifdef ESP32
const char* adapter_hostname = NULL;
tcpip_adapter_get_hostname(TCPIP_ADAPTER_IF_STA, &adapter_hostname);
if (adapter_hostname) {
memcpy(syslog->hostname, adapter_hostname, _POSIX_HOST_NAME_MAX);
}
else {
syslog->hostname[0] = '\0';
}
#else
gethostname(syslog->hostname, _POSIX_HOST_NAME_MAX + 1);
#endif
syslog->hostname[_POSIX_HOST_NAME_MAX] = '\0';
datagram_client_open(&syslog->client, host, port);
return &syslog->output;
}

1
src/r_api.c
View file

@ -29,6 +29,7 @@
#include "data_tag.h"
#include "list.h"
#include "optparse.h"
#include "output_file.h"
#include "output_mqtt.h"
#include "output_influx.h"
#include "write_sigrok.h"

2
tests/CMakeLists.txt
View file

@ -1,7 +1,7 @@
########################################################################
# Compile test cases
########################################################################
add_executable(data-test data-test.c)
add_executable(data-test data-test.c ../src/output_file.c)
target_link_libraries(data-test data)

1
tests/data-test.c
View file

@ -21,6 +21,7 @@
#include <stdio.h>
#include "data.h"
#include "output_file.h"
int main()
{

2
vs15/rtl_433.vcxproj
View file

@ -112,6 +112,7 @@ COPY ..\..\libusb\MS64\dll\libusb*.dll $(TargetDir)</Command>
<ClInclude Include="..\include\list.h" />
<ClInclude Include="..\include\mongoose.h" />
<ClInclude Include="..\include\optparse.h" />
<ClInclude Include="..\include\output_file.h" />
<ClInclude Include="..\include\output_influx.h" />
<ClInclude Include="..\include\output_mqtt.h" />
<ClInclude Include="..\include\pulse_analyzer.h" />
@ -150,6 +151,7 @@ COPY ..\..\libusb\MS64\dll\libusb*.dll $(TargetDir)</Command>
<ClCompile Include="..\src\list.c" />
<ClCompile Include="..\src\mongoose.c" />
<ClCompile Include="..\src\optparse.c" />
<ClCompile Include="..\src\output_file.c" />
<ClCompile Include="..\src\output_influx.c" />
<ClCompile Include="..\src\output_mqtt.c" />
<ClCompile Include="..\src\pulse_analyzer.c" />

6
vs15/rtl_433.vcxproj.filters
View file

@ -74,6 +74,9 @@
<ClInclude Include="..\include\optparse.h">
<Filter>Header Files</Filter>
</ClInclude>
<ClInclude Include="..\include\output_file.h">
<Filter>Header Files</Filter>
</ClInclude>
<ClInclude Include="..\include\output_influx.h">
<Filter>Header Files</Filter>
</ClInclude>
@ -184,6 +187,9 @@
<ClCompile Include="..\src\optparse.c">
<Filter>Source Files</Filter>
</ClCompile>
<ClCompile Include="..\src\output_file.c">
<Filter>Source Files</Filter>
</ClCompile>
<ClCompile Include="..\src\output_influx.c">
<Filter>Source Files</Filter>
</ClCompile>