Skip to content

Commit

Permalink
in_xbee: Periodic I/O sampling support
Browse files Browse the repository at this point in the history 
Signed-off-by: Takeshi HASEGAWA <hasegaw@gmail.com>
  • Loading branch information
@hasegaw
hasegaw committed Jul 2, 2015
1 parent d3fafc0 commit 3068ebe
Show file tree
Hide file tree
Showing 2 changed files with 128 additions and 13 deletions.
138 changes: 126 additions & 12 deletions plugins/in_xbee/in_xbee.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -170,9 +170,104 @@ void in_xbee_cb(struct xbee *xbee, struct xbee_con *con,
printf("\n");
#endif

if (! in_xbee_rx_queue_msgpack(ctx, (const char*) (*pkt)->data, (*pkt)->dataLen)) {
in_xbee_rx_queue_raw(ctx, (const char*) (*pkt)->data, (*pkt)->dataLen);
}

void in_xbee_io_sample_cb(struct xbee *xbee, struct xbee_con *con,
struct xbee_pkt **pkt, void **data)
{
struct flb_in_xbee_config *ctx;
int i;
int map_len = 0;
unsigned int mask_din, mask_ain;


if ((*pkt)->dataLen == 0) {
flb_debug("xbee data length too short, skip");
return;
}

ctx = *data;

unsigned char *p = (unsigned char*) (*pkt)->data;

if (*p != 1)
return;

mask_din = *(p + 1) << 8 | *(p + 2);
mask_ain = *(p + 3);

for (i = 0; i < 15; i++) {
if (mask_din & (1 << i))
map_len++;
if (mask_ain & (1 << i))
map_len++;
}

p += 4;

flb_debug("[xbee] IO sample: mask_din=0x%x mask_ain=%x map_len=%d", mask_din, mask_ain, map_len);

pthread_mutex_lock(&ctx->mtx_mp);

in_xbee_flush_if_needed(ctx);
ctx->buffer_id++;

msgpack_pack_array(&ctx->mp_pck, 2);
msgpack_pack_uint64(&ctx->mp_pck, time(NULL));
msgpack_pack_map(&ctx->mp_pck, map_len);

if (mask_din) {
/* sampled digital data sets */
int din = *p << 8 | *(p + 1);
p += 2;

for (i = 0; i < 15; i++) {
if (mask_din & (1 << i)) {
char name[6];
snprintf((char*) &name, sizeof(name), "DIO%d", i);

msgpack_pack_bin(&ctx->mp_pck, strlen((char*) &name));
msgpack_pack_bin_body(&ctx->mp_pck, (char*) &name, strlen((char*) &name));
msgpack_pack_int(&ctx->mp_pck, (din & (1 << i)) > 0);
}
}
}

if (mask_ain & 0x01) {
msgpack_pack_bin(&ctx->mp_pck, 3);
msgpack_pack_bin_body(&ctx->mp_pck, "AD0", 3);
msgpack_pack_int(&ctx->mp_pck, *p << 8 | *(p + 1));
p += 2;
}

if (mask_ain & 0x02) {
msgpack_pack_bin(&ctx->mp_pck, 3);
msgpack_pack_bin_body(&ctx->mp_pck, "AD1", 3);
msgpack_pack_int(&ctx->mp_pck, *p << 8 | *(p + 1));
p += 2;
}

if (mask_ain & 0x04) {
msgpack_pack_bin(&ctx->mp_pck, 3);
msgpack_pack_bin_body(&ctx->mp_pck, "AD2", 3);
msgpack_pack_int(&ctx->mp_pck, *p << 8 | *(p + 1));
p += 2;
}

if (mask_ain & 0x08) {
msgpack_pack_bin(&ctx->mp_pck, 3);
msgpack_pack_bin_body(&ctx->mp_pck, "AD3", 3);
msgpack_pack_int(&ctx->mp_pck, *p << 8 | *(p + 1));
p += 2;
}

if (mask_ain & 0x80) {
msgpack_pack_bin(&ctx->mp_pck, 3);
msgpack_pack_bin_body(&ctx->mp_pck, "VCC", 3);
msgpack_pack_int(&ctx->mp_pck, *p << 8 | *(p + 1));
p += 2;
}
pthread_mutex_unlock(&ctx->mtx_mp);
}

/* Callback triggered by timer */
Expand All @@ -183,7 +278,7 @@ int in_xbee_collect(struct flb_config *config, void *in_context)
(void) config;
struct flb_in_xbee_config *ctx = in_context;

if ((ret = xbee_conCallbackGet(ctx->con,
if ((ret = xbee_conCallbackGet(ctx->con_data,
(xbee_t_conCallback*) &p)) != XBEE_ENONE) {
flb_debug("xbee_conCallbackGet() returned: %d", ret);
return ret;
Expand Down Expand Up @@ -221,7 +316,7 @@ void *in_xbee_flush(void *in_context, int *size)
return buf;

fail:
pthread_mutex_lock(&ctx->mtx_mp);
pthread_mutex_unlock(&ctx->mtx_mp);
return NULL;
}

Expand All @@ -231,7 +326,6 @@ int in_xbee_init(struct flb_config *config)
int ret;
struct stat dev_st;
struct xbee *xbee;
struct xbee_con *con;
struct xbee_conAddress address;
struct flb_in_xbee_config *ctx;
struct xbee_conSettings settings;
Expand Down Expand Up @@ -277,6 +371,7 @@ int in_xbee_init(struct flb_config *config)

ctx->config = config;
pthread_mutex_init(&ctx->mtx_mp, NULL);
ctx->buffer_len = 0;

/* Init library */
xbee_init();
Expand All @@ -303,25 +398,44 @@ int in_xbee_init(struct flb_config *config)
xbee_logLevelSet(xbee, ctx->xbeeLogLevel);

/* Prepare a connection with the peer XBee */
if ((ret = xbee_conNew(xbee, &con, "Data", &address)) != XBEE_ENONE) {

if ((ret = xbee_conNew(xbee, &ctx->con_data, "Data", &address)) != XBEE_ENONE) {
xbee_log(xbee, -1, "xbee_conNew() returned: %d (%s)", ret, xbee_errorToStr(ret));
return ret;
}

xbee_conSettings(con, NULL, &settings);
xbee_conSettings(ctx->con_data, NULL, &settings);
settings.disableAck = ctx->xbeeDisableAck ? 1 : 0;
settings.catchAll = ctx->xbeeCatchAll ? 1 : 0;
xbee_conSettings(con, &settings, NULL);
xbee_conSettings(ctx->con_data, &settings, NULL);

ctx->con = con;
ctx->buffer_len = 0;
if ((ret = xbee_conDataSet(ctx->con_data, ctx, NULL)) != XBEE_ENONE) {
xbee_log(xbee, -1, "xbee_conDataSet() returned: %d", ret);
return ret;
}

if ((ret = xbee_conCallbackSet(ctx->con_data, in_xbee_cb, NULL)) != XBEE_ENONE) {
xbee_log(xbee, -1, "xbee_conCallbackSet() returned: %d", ret);
return ret;
}


if ((ret = xbee_conNew(xbee, &ctx->con_io, "I/O", &address)) != XBEE_ENONE) {
xbee_log(xbee, -1, "xbee_conNew() returned: %d (%s)", ret, xbee_errorToStr(ret));
return ret;
}

xbee_conSettings(ctx->con_io, NULL, &settings);
settings.disableAck = ctx->xbeeDisableAck ? 1 : 0;
settings.catchAll = ctx->xbeeCatchAll ? 1 : 0;
xbee_conSettings(ctx->con_io, &settings, NULL);

if ((ret = xbee_conDataSet(con, ctx, NULL)) != XBEE_ENONE) {
if ((ret = xbee_conDataSet(ctx->con_io, ctx, NULL)) != XBEE_ENONE) {
xbee_log(xbee, -1, "xbee_conDataSet() returned: %d", ret);
return ret;
}

if ((ret = xbee_conCallbackSet(con, in_xbee_cb, NULL)) != XBEE_ENONE) {
if ((ret = xbee_conCallbackSet(ctx->con_io, in_xbee_io_sample_cb, NULL)) != XBEE_ENONE) {
xbee_log(xbee, -1, "xbee_conCallbackSet() returned: %d", ret);
return ret;
}
Expand Down
3 changes: 2 additions & 1 deletion plugins/in_xbee/in_xbee_config.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,8 @@ struct flb_in_xbee_config {
char *xbeeMode;

/* Active connection context */
struct xbee_con *con;
struct xbee_con *con_data;
struct xbee_con *con_io;

/* buffering */
int buffer_len;
Expand Down

0 comments on commit 3068ebe

Please sign in to comment.