mirror/rtl_433
1
0
Fork 0
mirror of https://github.com/merbanan/rtl_433.git synced 2026-07-30 10:08:18 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add InfluxDB option for configurable prefix (#2586)

Browse Source
This commit is contained in:
Jan Kundrát
2026-07-07 12:47:31 +02:00
committed by GitHub
parent 5a3e53042e
commit 0ba8d5d21e
7 changed files with 188 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -530,6 +530,7 @@ E.g. -X "n=doorbell,m=OOK_PWM,s=400,l=800,r=7000,g=1000,match={24}0xa9878c,repea
Specify InfluxDB 2.0 server with e.g. -F "influx://localhost:9999/api/v2/write?org=<org>&bucket=<bucket>,token=<authtoken>"
Specify InfluxDB 1.x server with e.g. -F "influx://localhost:8086/write?db=<db>&p=<password>&u=<user>"
Additional parameter -M time:unix:usec:utc for correct timestamps in InfluxDB recommended
Additional parameter `metric` supports expanding keys (see the MQTT section just above for details).
[-F syslog[:[//]host[:port] (default: localhost:514)
Specify host/port for syslog with e.g. -F syslog:127.0.0.1:1514
[-F trigger:/path/to/file]

1
conf/rtl_433.example.conf
View File

@@ -203,6 +203,7 @@ report_meta protocol
# Specify InfluxDB 2.0 server with e.g. -F "influx://localhost:9999/api/v2/write?org=<org>&bucket=<bucket>,token=<authtoken>"
# Specify InfluxDB 1.x server with e.g. -F "influx://localhost:8086/write?db=<db>&p=<password>&u=<user>"
# Additional parameter -M time:unix:usec:utc for correct timestamps in InfluxDB recommended
# Additional parameter `metric` supports expanding keys (see the MQTT section just above for details).
# [-F syslog[:[//]host[:port] (default: localhost:514)
# Specify host/port for syslog with e.g. -F syslog:127.0.0.1:1514
# [-F trigger:/path/to/file]

16
include/string_expand.h Normal file
View File

@@ -0,0 +1,16 @@
/** @file
String formatting a-la MQTT topic
Copyright (C) 2019 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 "data.h"
typedef char *(expand_string_sanitizer)(char *, char *);
char *expand_topic_string(char *topic, char const *format, data_t *data, char const *hostname, expand_string_sanitizer sanitizer);

1
src/CMakeLists.txt
View File

@@ -46,6 +46,7 @@ add_library(r_433 STATIC
samp_grab.c
sdr.c
sigmf.c
string_expand.c
term_ctl.c
write_sigrok.c
devices/abmt.c

36
src/output_influx.c
View File

@@ -17,6 +17,7 @@
#include "logger.h"
#include "fatal.h"
#include "r_util.h"
#include "string_expand.h"
#include <stdlib.h>
#include <stdio.h>
@@ -38,6 +39,7 @@ typedef struct {
char hostname[64];
char url[400];
char extra_headers[150];
char metric_format[100];
tls_opts_t tls_opts;
int databufidxfill;
struct mbuf databufs[2];
@@ -120,10 +122,11 @@ static void influx_client_timer(struct mg_connection *nc, int ev, void *ev_data)
}
}
static influx_client_t *influx_client_init(influx_client_t *ctx, char const *url, char const *token)
static influx_client_t *influx_client_init(influx_client_t *ctx, char const *url, char const *token, const char *metric_format)
{
snprintf(ctx->url, sizeof(ctx->url), "%s", url);
snprintf(ctx->extra_headers, sizeof (ctx->extra_headers), "Authorization: Token %s\r\n", token);
snprintf(ctx->metric_format, sizeof(ctx->metric_format), "%s", metric_format);
return ctx;
}
@@ -339,7 +342,12 @@ static void R_API_CALLCONV print_influx_data(data_output_t *output, data_t *data
data_time = d;
}
if (!data_model) {
if (influx->metric_format[0]) {
char metric_name[1000];
mbuf_reserve(buf, sizeof (metric_name) + 1000);
expand_topic_string(metric_name, influx->metric_format, data, influx->hostname, influx_sanitize_tag);
mbuf_snprintf(buf, "%s", metric_name);
} else if (!data_model) {
// data isn't from device (maybe report for example)
// use hostname for measurement
@@ -347,7 +355,7 @@ static void R_API_CALLCONV print_influx_data(data_output_t *output, data_t *data
mbuf_snprintf(buf, "rtl_433_%s", influx->hostname);
}
else {
// use model for measurement
// default: use "model" for metric name
mbuf_reserve(buf, 1000);
str = &buf->buf[buf->len];
@@ -357,8 +365,17 @@ static void R_API_CALLCONV print_influx_data(data_output_t *output, data_t *data
// write tags
while (data) {
if (!strcmp(data->key, "model")
|| !strcmp(data->key, "time")) {
if ((!strcmp(data->key, "model") && strstr(influx->metric_format, "[model"))
|| (!strcmp(data->key, "type") && strstr(influx->metric_format, "[type"))
|| (!strcmp(data->key, "subtype") && strstr(influx->metric_format, "[subtype"))
|| (!strcmp(data->key, "hostname") && strstr(influx->metric_format, "[hostname"))
|| (!strcmp(data->key, "channel") && strstr(influx->metric_format, "[channel"))
|| (!strcmp(data->key, "id") && strstr(influx->metric_format, "[id"))
|| (!strcmp(data->key, "protocol") && strstr(influx->metric_format, "[protocol"))) {
// this field is already encoded in the metric name -> skip
} else if (!strcmp(data->key, "model") && !influx->metric_format[0]) {
// non-configurable metric format uses "model" for metric name -> skip
} else if (!strcmp(data->key, "time")) {
// skip
}
else if (!strcmp(data->key, "type")
@@ -486,6 +503,7 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts)
influx_sanitize_tag(influx->hostname, NULL);
char *token = NULL;
char *metric_format = NULL;
// param/opts starts with URL
if (!opts) {
@@ -527,6 +545,8 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts)
continue;
else if (!strcasecmp(key, "t") || !strcasecmp(key, "token"))
token = val;
else if (!strcasecmp(key, "metric"))
metric_format = val;
else if (!tls_param(&influx->tls_opts, key, val)) {
// ok
}
@@ -543,7 +563,9 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts)
influx->output.print_int = print_influx_int;
influx->output.output_free = data_output_influx_free;
print_logf(LOG_CRITICAL, "InfluxDB", "Publishing data to InfluxDB (%s)", url);
print_logf(LOG_CRITICAL, "InfluxDB", "Publishing data to InfluxDB (%s, %s%s)", url,
metric_format ? "dynamic metric " : "static metric",
metric_format ? metric_format : "");
influx->mgr = mgr;
@@ -551,7 +573,7 @@ struct data_output *data_output_influx_create(struct mg_mgr *mgr, char *opts)
struct mg_add_sock_opts timer_opts = {.user_data = influx};
influx->timer = mg_add_sock_opt(mgr, INVALID_SOCKET, influx_client_timer, timer_opts);
influx_client_init(influx, url, token);
influx_client_init(influx, url, token, metric_format);
return (struct data_output *)influx;
}

127
src/output_mqtt.c
View File

@@ -16,6 +16,7 @@
#include "logger.h"
#include "fatal.h"
#include "r_util.h"
#include "string_expand.h"
#include <stdlib.h>
#include <stdio.h>
@@ -447,9 +448,9 @@ static void mqtt_client_free(mqtt_client_t *ctx)
/* Helper */
/// clean the topic inplace to [-.A-Za-z0-9], esp. not whitespace, +, #, /, $
static char *mqtt_sanitize_topic(char *topic)
static char *mqtt_sanitize_topic(char *topic, char *end)
{
for (char *p = topic; *p; ++p)
for (char *p = topic; p < end && *p; ++p)
if (*p != '-' && *p != '.' && (*p < 'A' || *p > 'Z') && (*p < 'a' || *p > 'z') && (*p < '0' || *p > '9'))
*p = '_';
@@ -484,122 +485,6 @@ static void R_API_CALLCONV print_mqtt_array(data_output_t *output, data_array_t
*orig = '\0'; // restore topic
}
static char *append_topic(char *topic, data_t *data)
{
if (data->type == DATA_STRING) {
strcpy(topic, data->value.v_ptr); // NOLINT
mqtt_sanitize_topic(topic);
topic += strlen(data->value.v_ptr);
}
else if (data->type == DATA_INT) {
topic += sprintf(topic, "%d", data->value.v_int);
}
else {
print_logf(LOG_ERROR, __func__, "Can't append data type %d to topic", data->type);
}
return topic;
}
static char *expand_topic(char *topic, char const *format, data_t *data, char const *hostname)
{
// collect well-known top level keys
data_t *data_type = NULL;
data_t *data_model = NULL;
data_t *data_subtype = NULL;
data_t *data_channel = NULL;
data_t *data_id = NULL;
data_t *data_protocol = NULL;
for (data_t *d = data; d; d = d->next) {
if (!strcmp(d->key, "type"))
data_type = d;
else if (!strcmp(d->key, "model"))
data_model = d;
else if (!strcmp(d->key, "subtype"))
data_subtype = d;
else if (!strcmp(d->key, "channel"))
data_channel = d;
else if (!strcmp(d->key, "id"))
data_id = d;
else if (!strcmp(d->key, "protocol")) // NOTE: needs "-M protocol"
data_protocol = d;
}
// consume entire format string
while (format && *format) {
data_t *data_token = NULL;
char const *string_token = NULL;
int leading_slash = 0;
char const *t_start = NULL;
char const *t_end = NULL;
char const *d_start = NULL;
char const *d_end = NULL;
// copy until '['
while (*format && *format != '[')
*topic++ = *format++;
// skip '['
if (!*format)
break;
++format;
// read slash
if (!leading_slash && (*format < 'a' || *format > 'z')) {
leading_slash = *format;
format++;
}
// read key until : or ]
t_start = t_end = format;
while (*format && *format != ':' && *format != ']' && *format != '[')
t_end = ++format;
// read default until ]
if (*format == ':') {
d_start = d_end = ++format;
while (*format && *format != ']' && *format != '[')
d_end = ++format;
}
// check for proper closing
if (*format != ']') {
print_log(LOG_FATAL, __func__, "unterminated token");
exit(1);
}
++format;
// resolve token
if (!strncmp(t_start, "hostname", t_end - t_start))
string_token = hostname;
else if (!strncmp(t_start, "type", t_end - t_start))
data_token = data_type;
else if (!strncmp(t_start, "model", t_end - t_start))
data_token = data_model;
else if (!strncmp(t_start, "subtype", t_end - t_start))
data_token = data_subtype;
else if (!strncmp(t_start, "channel", t_end - t_start))
data_token = data_channel;
else if (!strncmp(t_start, "id", t_end - t_start))
data_token = data_id;
else if (!strncmp(t_start, "protocol", t_end - t_start))
data_token = data_protocol;
else {
print_logf(LOG_FATAL, __func__, "unknown token \"%.*s\"", (int)(t_end - t_start), t_start);
exit(1);
}
// append token or default
if (!data_token && !string_token && !d_start)
continue;
if (leading_slash)
*topic++ = leading_slash;
if (data_token)
topic = append_topic(topic, data_token);
else if (string_token)
topic += sprintf(topic, "%s", string_token);
else
topic += sprintf(topic, "%.*s", (int)(d_end - d_start), d_start);
}
*topic = '\0';
return topic;
}
// <prefix>[/type][/model][/subtype][/channel][/id]/battery: "OK"|"LOW"
static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data, char const *format)
{
@@ -628,7 +513,7 @@ static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data,
return; // NOTE: skip output on alloc failure.
}
data_print_jsons(data, message, message_size);
expand_topic(mqtt->topic, mqtt->states, data, mqtt->hostname);
expand_topic_string(mqtt->topic, mqtt->states, data, mqtt->hostname, mqtt_sanitize_topic);
mqtt_client_publish(mqtt->mqc, mqtt->topic, message);
*mqtt->topic = '\0'; // clear topic
free(message);
@@ -640,7 +525,7 @@ static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data,
if (mqtt->events) {
char message[2048]; // we expect the biggest strings to be around 500 bytes.
data_print_jsons(data, message, sizeof(message));
expand_topic(mqtt->topic, mqtt->events, data, mqtt->hostname);
expand_topic_string(mqtt->topic, mqtt->events, data, mqtt->hostname, mqtt_sanitize_topic);
mqtt_client_publish(mqtt->mqc, mqtt->topic, message);
*mqtt->topic = '\0'; // clear topic
}
@@ -650,7 +535,7 @@ static void R_API_CALLCONV print_mqtt_data(data_output_t *output, data_t *data,
return;
}
end = expand_topic(mqtt->topic, mqtt->devices, data, mqtt->hostname);
end = expand_topic_string(mqtt->topic, mqtt->devices, data, mqtt->hostname, mqtt_sanitize_topic);
}
while (data) {

134
src/string_expand.c Normal file
View File

@@ -0,0 +1,134 @@
/** @file
String formatting a-la MQTT topic
Copyright (C) 2019 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 "string_expand.h"
#include "logger.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
static char *append_topic(char *topic, data_t *data, expand_string_sanitizer sanitizer)
{
if (data->type == DATA_STRING) {
strcpy(topic, data->value.v_ptr); // NOLINT
size_t len = strlen(topic);
(*sanitizer)(topic, topic + len);
topic += len;
}
else if (data->type == DATA_INT) {
topic += sprintf(topic, "%d", data->value.v_int);
}
else {
print_logf(LOG_ERROR, __func__, "Can't append data type %d to topic", data->type);
}
return topic;
}
char *expand_topic_string(char *topic, char const *format, data_t *data, char const *hostname, expand_string_sanitizer sanitizer)
{
// collect well-known top level keys
data_t *data_type = NULL;
data_t *data_model = NULL;
data_t *data_subtype = NULL;
data_t *data_channel = NULL;
data_t *data_id = NULL;
data_t *data_protocol = NULL;
for (data_t *d = data; d; d = d->next) {
if (!strcmp(d->key, "type"))
data_type = d;
else if (!strcmp(d->key, "model"))
data_model = d;
else if (!strcmp(d->key, "subtype"))
data_subtype = d;
else if (!strcmp(d->key, "channel"))
data_channel = d;
else if (!strcmp(d->key, "id"))
data_id = d;
else if (!strcmp(d->key, "protocol")) // NOTE: needs "-M protocol"
data_protocol = d;
}
// consume entire format string
while (format && *format) {
data_t *data_token = NULL;
char const *string_token = NULL;
int leading_slash = 0;
char const *t_start = NULL;
char const *t_end = NULL;
char const *d_start = NULL;
char const *d_end = NULL;
// copy until '['
while (*format && *format != '[')
*topic++ = *format++;
// skip '['
if (!*format)
break;
++format;
// read slash
if (!leading_slash && (*format < 'a' || *format > 'z')) {
leading_slash = *format;
format++;
}
// read key until : or ]
t_start = t_end = format;
while (*format && *format != ':' && *format != ']' && *format != '[')
t_end = ++format;
// read default until ]
if (*format == ':') {
d_start = d_end = ++format;
while (*format && *format != ']' && *format != '[')
d_end = ++format;
}
// check for proper closing
if (*format != ']') {
print_log(LOG_FATAL, __func__, "unterminated token");
exit(1);
}
++format;
// resolve token
if (!strncmp(t_start, "hostname", t_end - t_start))
string_token = hostname;
else if (!strncmp(t_start, "type", t_end - t_start))
data_token = data_type;
else if (!strncmp(t_start, "model", t_end - t_start))
data_token = data_model;
else if (!strncmp(t_start, "subtype", t_end - t_start))
data_token = data_subtype;
else if (!strncmp(t_start, "channel", t_end - t_start))
data_token = data_channel;
else if (!strncmp(t_start, "id", t_end - t_start))
data_token = data_id;
else if (!strncmp(t_start, "protocol", t_end - t_start))
data_token = data_protocol;
else {
print_logf(LOG_FATAL, __func__, "unknown token \"%.*s\"", (int)(t_end - t_start), t_start);
exit(1);
}
// append token or default
if (!data_token && !string_token && !d_start)
continue;
if (leading_slash)
*topic++ = leading_slash;
if (data_token)
topic = append_topic(topic, data_token, sanitizer);
else if (string_token)
topic += sprintf(topic, "%s", string_token);
else
topic += sprintf(topic, "%.*s", (int)(d_end - d_start), d_start);
}
*topic = '\0';
return topic;
}