Golioth stream batch data demo for nrf9160dk. Use Golioth Hello example and replace main.c with this.

main.c

/*
 * Copyright (c) 2024 Golioth, Inc.
 *
 * SPDX-License-Identifier: Apache-2.0
 */

#include <zephyr/logging/log.h>
LOG_MODULE_REGISTER(golioth_batch, LOG_LEVEL_DBG);

#include <golioth/client.h>
#include <golioth/stream.h>
#include <samples/common/sample_credentials.h>
#include <string.h>
#include <zephyr/kernel.h>
#include <zephyr/sys/timeutil.h>
#include <nrf_modem_at.h>
#include <time.h>
#include <zcbor_encode.h>

#include <samples/common/net_connect.h>

#define UPTIME_UNIT_PER_S 1000
#define KEY_MAX_SIZE 10
#define STREAM_BUF_SIZE 3

#define JSON_MEMBER_FMT "{\"ts\":%llu,\"%s\":%u}"
#define MAX_TS_STR_LEN 20
#define MAX_VAL_STR_LEN 10

struct reading {
    time_t ts;
    char key[KEY_MAX_SIZE];
    int val;
};

static struct reading stream_buf[STREAM_BUF_SIZE];

static struct golioth_client *client;
static K_SEM_DEFINE(connected, 0, 1);

static void on_client_event(struct golioth_client *client,
                            enum golioth_client_event event,
                            void *arg)
{
    bool is_connected = (event == GOLIOTH_CLIENT_EVENT_CONNECTED);
    if (is_connected)
    {
        k_sem_give(&connected);
    }
    LOG_INF("Golioth client %s", is_connected ? "connected" : "disconnected");
}

static void async_push_handler(struct golioth_client *client,
                               const struct golioth_response *response,
                               const char *path,
                               void *arg)
{
    if (response->status != GOLIOTH_OK)
    {
        LOG_WRN("Failed to push cached data: %d", response->status);
        return;
    }

    LOG_DBG("Cached data successfully pushed");
}

/* Not used in this example but this produces the CBOR equivalent of cached data */
/* TODO: set buf[] size based on #define values from top of file */
static int push_cbor_buffer_to_golioth(struct reading *s_buf, size_t tot_members)
{
    bool ok;
    uint8_t buf[256];
    ZCBOR_STATE_E(encode_state, 1, buf, sizeof(buf), 1);

    ok = zcbor_list_start_encode(encode_state, tot_members);

    if (!ok)
    {
        LOG_ERR("Error starting CBOR encoding");
        return ok;
    }

    for (int i = 0; i < tot_members; i++)
    {
        ok = zcbor_map_start_encode(encode_state, 2) &&
             zcbor_tstr_put_lit(encode_state, "ts") &&
             zcbor_uint64_put(encode_state, s_buf[i].ts) &&
             zcbor_tstr_put_term(encode_state, s_buf[i].key) &&
             zcbor_int32_put(encode_state, s_buf[i].val) &&
             zcbor_map_end_encode(encode_state, 2);

        if (!ok)
        {
            LOG_ERR("CBOR encoding error on member #%d", i);
            return ok;
        }
    }

    ok = zcbor_list_end_encode(encode_state, tot_members);

    size_t payload_size = encode_state->payload - buf;
    LOG_HEXDUMP_DBG(buf, payload_size, "cbor");

    int err = golioth_stream_set_async(client,
                                       "",
                                       GOLIOTH_CONTENT_TYPE_CBOR,
                                       buf,
                                       payload_size,
                                       async_push_handler,
                                       NULL);

    return err;
}

static int push_json_buffer_to_golioth(struct reading *s_buf, size_t tot_members)
{
    /* Calculate size of each member push 32 bytes of headroom */
    uint8_t buf[((strlen(JSON_MEMBER_FMT) + KEY_MAX_SIZE + MAX_TS_STR_LEN + MAX_VAL_STR_LEN)
                 * STREAM_BUF_SIZE)
                + 32];
    int idx = 1;

    memset(buf, 0, sizeof(buf));
    buf[0] = '[';

    for (int i = 0; i < tot_members; i++)
    {
        idx = strlen(buf);
        snprintk(buf + idx,
                 sizeof(buf) - idx,
                 JSON_MEMBER_FMT,
                 s_buf[i].ts,
                 s_buf[i].key,
                 s_buf[i].val);

        if (i < tot_members - 1)
        {
            buf[strlen(buf)] = ',';
        }
    }

    buf[strlen(buf)] = ']';

    LOG_HEXDUMP_DBG(buf, strlen(buf), "json");

    int err = golioth_stream_set_async(client,
                                       "",
                                       GOLIOTH_CONTENT_TYPE_JSON,
                                       buf,
                                       strlen(buf),
                                       async_push_handler,
                                       NULL);
    return err;
}

/* This is an overly simple way of getting time and only works for nrf91 */
static time_t get_epoch(void)
{
    struct tm source_time = { 0 };

    int err = nrf_modem_at_scanf("AT+CCLK?", "+CCLK: \"%d/%d/%d,%d:%d:%d-%d\"",
                                 &source_time.tm_year, &source_time.tm_mon, &source_time.tm_mday,
                                 &source_time.tm_hour, &source_time.tm_min, &source_time.tm_min,
                                 &source_time.tm_sec);
    if (err < 0) {
        LOG_ERR("Error parsing time: %d", err);
    } else {
        /* Adjust for tm format */
        source_time.tm_year += 100;
        source_time.tm_mon -= 1;

        time_t dest_time = timeutil_timegm(&source_time);
        LOG_DBG("Epoch: %jd", (intmax_t) dest_time);

        return dest_time;
    }

    LOG_ERR("Unable to get time. Beware: Your timestamps will start at 1/1/1970!!");
    return 0;
}

int main(void)
{
    int counter = 0;

    LOG_DBG("start batch sample");

    net_connect();

    const struct golioth_client_config *client_config = golioth_sample_credentials_get();

    client = golioth_client_create(client_config);

    golioth_client_register_event_callback(client, on_client_event, NULL);

    k_sem_take(&connected, K_FOREVER);

    time_t epoch = get_epoch();
    if (epoch == 0)
    {
    }
    int buf_idx = 0;

    while (true)
    {
        uint64_t ts = epoch + (k_uptime_get() / UPTIME_UNIT_PER_S);
        LOG_INF("Caching hello! %d, %llu", counter, ts);

        /* Cache the reading */
        stream_buf[buf_idx].ts = ts;
        snprintk(stream_buf[buf_idx].key, KEY_MAX_SIZE, "%s", "counter");
        stream_buf[buf_idx].val = counter;
        ++buf_idx;

        /* Check if buffer is full */
        if (buf_idx >= STREAM_BUF_SIZE) {
            buf_idx = 0;

            /* push all data in buffer to Golioth */
            push_json_buffer_to_golioth(stream_buf, STREAM_BUF_SIZE);

            /* resync epoch with cell network */
            epoch = get_epoch();
        }

        ++counter;
        k_sleep(K_SECONDS(5));
    }

    return 0;
}