modules: Add piping for decoding incoming CBOR shadow data

Add piping for decoding incoming CBOR shadow data

Author: simensrostad

app/CMakeLists.txt

@@ -21,6 +21,7 @@ add_subdirectory(src/modules/cloud)
 add_subdirectory(src/modules/location)
 add_subdirectory(src/modules/fota)
 add_subdirectory(src/modules/button)
+add_subdirectory(src/cbor)
 
 # Optional modules
 add_subdirectory_ifdef(CONFIG_APP_POWER src/modules/power)

app/prj.conf

@@ -210,3 +210,6 @@ CONFIG_TASK_WDT_MIN_TIMEOUT=10000
 # Device power management
 CONFIG_PM_DEVICE=y
 CONFIG_PM_DEVICE_SHELL=y
+
+# JSON library
+CONFIG_JSON_LIBRARY=y

app/src/Kconfig.main

@@ -16,10 +16,6 @@ config APP_MODULE_THREAD_STACK_SIZE
 	int "Thread stack size"
 	default 3200
 
-config APP_MODULE_RECV_BUFFER_SIZE
-	int "Receive buffer size"
-	default 1024
-
 config APP_REQUEST_NETWORK_QUALITY
 	bool "Request network quality"
 	help

app/src/cbor/CMakeLists.txt

@@ -0,0 +1,8 @@
+#
+# Copyright (c) 2025 Nordic Semiconductor ASA
+#
+# SPDX-License-Identifier: LicenseRef-Nordic-5-Clause
+#
+
+target_sources(app PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/cbor_helper.c)
+target_include_directories(app PRIVATE .)

app/src/cbor/cbor_helper.c

@@ -0,0 +1,12 @@
+#include <zephyr/logging/log.h>
+#include <zcbor_decode.h>
+
+LOG_MODULE_DECLARE(main, CONFIG_APP_LOG_LEVEL);
+
+int get_update_interval_from_cbor_response(const uint8_t *cbor,
+		       size_t len,
+		       uint64_t *interval_sec)
+{
+	interval_sec = 60;
+	return 0;
+}

app/src/cbor/cbor_helper.h

@@ -0,0 +1,5 @@
+#include <zephyr/types.h>
+
+int get_update_interval_from_cbor_response(const uint8_t *cbor,
+					   size_t len,
+					   uint64_t *interval_sec);

app/src/cbor/device_shadow.cddl

@@ -0,0 +1,53 @@
+; Define the basic structure of the JSON object
+
+; The 'any' type is used in this schema to allow forward compatibility.
+; They are inserted into maps that may contain additional fields in the future.
+; This prevents cbor decoding from failing in the field when the cloud side has new fields.
+app-object = {
+    ? "nrfcloud_mqtt_topic_prefix": tstr,
+    ? "pairing": pairing-type,
+    ? "lwm2m": lwm2m-map,
+    * tstr => any
+}
+
+; Define the pairing object with nested topics
+pairing-type = {
+    "state": tstr,
+    "topics": {
+        "d2c": tstr,
+        "c2d": tstr,
+        * tstr => any
+    },
+    * any => any
+}
+
+lwm2m-map = {
+  ? "14240:1.0": led,
+  ? "14301:1.0": config,
+  * tstr => any
+}
+
+led = {
+  "0": led_inner_object,
+  * tstr => any
+}
+
+led_inner_object = {
+  ? "0": int .size 4,
+  ? "1": int .size 4,
+  ? "2": int .size 4,
+  "99": int .size 8,
+  * tstr => any
+}
+
+config = {
+  "0": config_inner_object,
+  * tstr => any
+}
+
+config_inner_object = {
+  ? "0": int .size 8,
+  ? "1": bool,
+  "99": int .size 8,
+  * tstr => any
+}

app/src/main.c

@@ -16,6 +16,7 @@
 #include "cloud_module.h"
 #include "fota.h"
 #include "location.h"
+#include "cbor_helper.h"
 
 #if defined(CONFIG_APP_LED)
 #include "led.h"
@@ -35,15 +36,14 @@
 /* Register log module */
 LOG_MODULE_REGISTER(main, CONFIG_APP_LOG_LEVEL);
 
-#define MAX_MSG_SIZE	(MAX(sizeof(struct cloud_shadow_response),				\
-			 MAX(sizeof(struct cloud_payload),					\
+#define MAX_MSG_SIZE	(MAX(sizeof(struct cloud_msg),						\
 			 /* Button channel payload size */					\
 			 MAX(sizeof(uint8_t),							\
 			 /* Timer channel payload size */					\
 			 MAX(sizeof(int),							\
 			 MAX(sizeof(enum fota_msg_type),					\
 			 MAX(sizeof(enum location_msg_type),					\
-			 MAX(sizeof(struct network_msg), POWER_MSG_SIZE))))))))
+			 MAX(sizeof(struct network_msg), POWER_MSG_SIZE)))))))
 
 /* Register subscriber */
 ZBUS_MSG_SUBSCRIBER_DEFINE(main_subscriber);
@@ -232,6 +232,7 @@ static const struct smf_state states[] = {
 };
 
 /* Static helper function */
+
 static void task_wdt_callback(int channel_id, void *user_data)
 {
 	LOG_ERR("Watchdog expired, Channel: %d, Thread: %s",
@@ -441,6 +442,7 @@ static void triggering_entry(void *o)
 
 static void triggering_run(void *o)
 {
+	int err;
 	struct main_state *state_object = o;
 
 	if (state_object->chan == &CLOUD_CHAN) {
@@ -454,13 +456,19 @@ static void triggering_run(void *o)
 		}
 
 		if (msg.type == CLOUD_SHADOW_RESPONSE) {
-			/* Missing: Parse the interval received in the shadow response,
-			 * write to state object and schedule new interval
-			 */
+			err = get_update_interval_from_cbor_response(msg.response.buffer,
+								     msg.response.buffer_data_len,
+								     &state_object->interval_sec);
+			if (err) {
+				LOG_ERR("json_parse, error: %d", err);
+				SEND_FATAL_ERROR();
+				return;
+			}
 
-			int err = k_work_reschedule(&trigger_work,
-						    K_SECONDS(state_object->interval_sec));
+			LOG_WRN("Received new interval: %lld seconds", state_object->interval_sec);
 
+			err = k_work_reschedule(&trigger_work,
+						K_SECONDS(state_object->interval_sec));
 			if (err < 0) {
 				LOG_ERR("k_work_reschedule, error: %d", err);
 				SEND_FATAL_ERROR();

app/src/modules/cloud/cloud_module.c

@@ -11,12 +11,9 @@
 #include <zephyr/task_wdt/task_wdt.h>
 #include <net/nrf_cloud.h>
 #include <net/nrf_cloud_coap.h>
+#include <nrf_cloud_coap_transport.h>
 #include <app_version.h>
 
-#if defined(CONFIG_MEMFAULT)
-#include <memfault/core/trace_event.h>
-#endif /* CONFIG_MEMFAULT */
-
 #include "cloud_module.h"
 #include "app_common.h"
 #include "network.h"
@@ -47,7 +44,7 @@ LOG_MODULE_REGISTER(cloud, CONFIG_APP_CLOUD_LOG_LEVEL);
 #define ENV_MSG_SIZE	0
 #endif /* CONFIG_APP_ENVIRONMENTAL) */
 
-#define MAX_MSG_SIZE	(MAX(sizeof(struct cloud_payload),					\
+#define MAX_MSG_SIZE	(MAX(sizeof(struct cloud_msg),						\
 			 MAX(sizeof(struct network_msg),					\
 			 MAX(BAT_MSG_SIZE, ENV_MSG_SIZE))))
 
@@ -467,13 +464,19 @@ static void state_connected_exit(void *o)
 static void shadow_get(bool delta_only)
 {
 	int err;
-	uint8_t recv_buf[CONFIG_APP_MODULE_RECV_BUFFER_SIZE] = { 0 };
-	size_t recv_buf_len = sizeof(recv_buf);
+	struct cloud_msg msg = {
+		.type = CLOUD_SHADOW_RESPONSE,
+		.response = {
+			.buffer_data_len = sizeof(msg.response.buffer),
+		},
+	};
 
 	LOG_DBG("Requesting device shadow from the device");
 
-	err = nrf_cloud_coap_shadow_get(recv_buf, &recv_buf_len, delta_only,
-					COAP_CONTENT_FORMAT_APP_JSON);
+	err = nrf_cloud_coap_shadow_get(msg.response.buffer,
+					&msg.response.buffer_data_len,
+					delta_only,
+					COAP_CONTENT_FORMAT_APP_CBOR);
 	if (err == -EACCES) {
 		LOG_WRN("Not connected, error: %d", err);
 		return;
@@ -486,12 +489,46 @@ static void shadow_get(bool delta_only)
 	} else if (err > 0) {
 		LOG_WRN("Cloud error: %d", err);
 		return;
+	} else if (err == -E2BIG) {
+		LOG_WRN("The provided buffer is not large enough, error: %d", err);
+		return;
 	} else if (err) {
 		LOG_ERR("Failed to request shadow delta: %d", err);
 		return;
 	}
 
-	/* No further processing of shadow is implemented */
+	if (msg.response.buffer_data_len == 0) {
+		LOG_DBG("No shadow delta changes available");
+		return;
+	}
+
+	/* Workaroud: Sometimes nrf_cloud_coap_shadow_get() returns 0 even though obtaining
+	 * the shadow failed. Ignore the payload if the first 10 bytes are zero.
+	 */
+	if (!memcmp(msg.response.buffer, "\0\0\0\0\0\0\0\0\0\0", 10)) {
+		LOG_WRN("Returned buffeør is empty, ignore");
+		return;
+	}
+
+	err = zbus_chan_pub(&CLOUD_CHAN, &msg, K_SECONDS(1));
+	if (err) {
+		LOG_ERR("zbus_chan_pub, error: %d", err);
+		SEND_FATAL_ERROR();
+		return;
+	}
+
+	/* Clear the shadow delta by reporting the same data back to the shadow reported state  */
+	err = nrf_cloud_coap_patch("state/reported", NULL,
+				   msg.response.buffer,
+				   msg.response.buffer_data_len,
+				   COAP_CONTENT_FORMAT_APP_CBOR,
+				   true,
+				   NULL,
+				   NULL);
+	if (err) {
+		LOG_ERR("Failed to patch the device shadow, error: %d", err);
+		return;
+	}
 }
 
 static void state_connected_ready_entry(void *o)
@@ -641,9 +678,9 @@ static void state_connected_ready_run(void *o)
 #endif /* CONFIG_APP_ENVIRONMENTAL */
 
 	if (state_object->chan == &PAYLOAD_CHAN) {
-		const struct cloud_payload *payload = MSG_TO_PAYLOAD(state_object->msg_buf);
+		const struct cloud_msg msg = MSG_TO_CLOUD_MSG(state_object->msg_buf);
 
-		err = nrf_cloud_coap_json_message_send(payload->buffer, false, confirmable);
+		err = nrf_cloud_coap_json_message_send(msg.payload.buffer, false, confirmable);
 		if (err == -ENETUNREACH) {
 			LOG_WRN("Network is unreachable, error: %d", err);
 			return;

app/src/modules/cloud/cloud_module.h

@@ -22,12 +22,14 @@ ZBUS_CHAN_DECLARE(
 
 struct cloud_payload {
 	uint8_t buffer[CONFIG_APP_CLOUD_PAYLOAD_BUFFER_MAX_SIZE];
-	size_t buffer_len;
+	/* Length of the data stored inside the buffer */
+	size_t buffer_data_len;
 };
 
 struct cloud_shadow_response {
 	uint8_t buffer[CONFIG_APP_CLOUD_SHADOW_RESPONSE_BUFFER_MAX_SIZE];
-	size_t buffer_len;
+	/* Length of the data stored inside the buffer */
+	size_t buffer_data_len;
 };
 
 enum cloud_msg_type {
@@ -43,11 +45,10 @@ enum cloud_msg_type {
 struct cloud_msg {
 	enum cloud_msg_type type;
 	struct cloud_payload payload;
+	struct cloud_shadow_response response;
 };
 
 #define MSG_TO_CLOUD_MSG(_msg)	(*(const struct cloud_msg *)_msg)
-#define MSG_TO_PAYLOAD(_msg)	((struct cloud_payload *)_msg)
-#define MSG_TO_SHADOW_RESPONSE(_msg)	(*(const struct cloud_shadow_response *)_msg)
 
 #ifdef __cplusplus
 }

app/src/modules/cloud/cloud_module_shell.c

@@ -47,10 +47,10 @@ static int cmd_publish(const struct shell *sh, size_t argc, char **argv)
 		return 1;
 	}
 
-	payload.buffer_len = err;
+	payload.buffer_data_len = err;
 
 	(void)shell_print(sh, "Sending on payload channel: %s (%d bytes)",
-			  payload.buffer, payload.buffer_len);
+			  payload.buffer, payload.buffer_data_len);
 
 	err = zbus_chan_pub(&PAYLOAD_CHAN, &payload, K_SECONDS(1));
 	if (err) {

tests/module/cloud/CMakeLists.txt

@@ -45,6 +45,5 @@ target_compile_definitions(app PRIVATE
 	-DCONFIG_LTE_LC_CONN_EVAL_MODULE=1
 	-DCONFIG_LTE_LC_EDRX_MODULE=1
 	-DCONFIG_LTE_LC_PSM_MODULE=1
-	-DCONFIG_APP_MODULE_RECV_BUFFER_SIZE=1024
 	-DCOAP_CONTENT_FORMAT_APP_JSON=50
 )

tests/module/main/CMakeLists.txt

@@ -60,7 +60,6 @@ target_compile_definitions(app PRIVATE
 	-DCONFIG_LTE_LC_CONN_EVAL_MODULE=1
 	-DCONFIG_LTE_LC_EDRX_MODULE=1
 	-DCONFIG_LTE_LC_PSM_MODULE=1
-	-DCONFIG_APP_MODULE_RECV_BUFFER_SIZE=1024
 	-DCOAP_CONTENT_FORMAT_APP_JSON=50
 	-DCONFIG_APP_POWER=1
 	-DCONFIG_APP_ENVIRONMENTAL=1

tests/module/main/src/main.c

@@ -141,19 +141,17 @@ static void send_location_search_done(void)
 	TEST_ASSERT_EQUAL(0, err);
 }
 
-/*
 static void send_config(uint64_t interval)
 {
-	const struct configuration config = {
-		.update_interval_present = true,
-		.update_interval = interval,
+	const struct cloud_msg msg = {
+		.type = CLOUD_SHADOW_RESPONSE,
+		.response.buffer_len = "sometinhg"
 	};
 
-	int err = zbus_chan_pub(&CONFIG_CHAN, &config, K_SECONDS(1));
+	int err = zbus_chan_pub(&CLOUD_CHAN, &shadow_response, K_SECONDS(1));
 
 	TEST_ASSERT_EQUAL(0, err);
 }
-*/
 
 static void send_cloud_disconnected(void)
 {
@@ -214,7 +212,6 @@ void test_button_press_on_disconnected(void)
 	check_no_events(7200);
 }
 
-/*
 void test_trigger_interval_change_in_connected(void)
 {
 	send_cloud_connected_ready_to_send();
@@ -230,9 +227,8 @@ void test_trigger_interval_change_in_connected(void)
 	send_cloud_disconnected();
 	check_no_events(WEEK_IN_SECONDS);
 }
-*/
 
-/*
+
 void test_trigger_disconnect_and_connect_when_triggering(void)
 {
 	send_cloud_connected_ready_to_send();
@@ -255,7 +251,6 @@ void test_trigger_disconnect_and_connect_when_triggering(void)
 	send_cloud_disconnected();
 	check_no_events(WEEK_IN_SECONDS);
 }
-*/
 
 /* This is required to be added to each test. That is because unity's
  * main may return nonzero, while zephyr's main currently must