caf: modules: sensor_data_aggregator: changed buffering

Handling aggregated events on receiving side required using DTS to
determine data layout. Now sensor_data_aggregator_event is using
struct sensor_value data buffer and carries number of sensor_values
in a single sample, which is sufficient to describe data layout.
caf_sensor_manager sample, tests, yaml and overlays has been updated
to account for new data structure.

JIRA: NCSDK-18421

Signed-off-by: Mateusz Michalek <[email protected]>

dts/bindings/caf/caf,aggregator.yaml

@@ -18,10 +18,10 @@ properties:
     type: int
     default: 120
 
-  sensor_data_size:
-    description: size in bytes of single sensor data, range 1-255.
+  sample_size:
+    description: number of sensor_values in a single sample, range 1-31.
     type: int
-    default: 4
+    default: 1
 
   buf_count:
     description: Number of buffers in aggregator, range 1-255.

include/caf/events/sensor_data_aggregator_event.h

@@ -27,9 +27,10 @@ extern "C" {
 struct sensor_data_aggregator_event {
 	struct app_event_header header;
 	const char *sensor_descr;
-	uint8_t *buf;
+	struct sensor_value *samples;
 	enum sensor_state sensor_state;
 	uint8_t sample_cnt;
+	uint8_t values_in_sample;
 };
 
 /** @brief Sensor data aggregator release buffer event.
@@ -38,7 +39,7 @@ struct sensor_data_aggregator_event {
  */
 struct sensor_data_aggregator_release_buffer_event {
 	struct app_event_header header;
-	uint8_t *buf;
+	struct sensor_value *samples;
 	const char *sensor_descr;
 };
 

samples/caf_sensor_manager/app.overlay

@@ -9,7 +9,7 @@
 		compatible = "caf,aggregator";
 		sensor_descr = "accel_sim_xyz";
 		buf_data_length = <240>;
-		sensor_data_size = <24>;
+		sample_size= <3>;
 		status = "okay";
 	};
 };

samples/caf_sensor_manager/boards/nrf52840dk_nrf52840.overlay

@@ -15,7 +15,7 @@
 		compatible = "caf,aggregator";
 		sensor_descr = "accel_sim_xyz";
 		buf_data_length = <240>;
-		sensor_data_size = <24>;
+		sample_size = <3>;
 		status = "okay";
 	};
 };

samples/caf_sensor_manager/boards/nrf5340dk_nrf5340_cpuapp_singlecore.overlay

@@ -9,7 +9,7 @@
 		compatible = "caf,aggregator";
 		sensor_descr = "accel_sim_xyz";
 		buf_data_length = <240>;
-		sensor_data_size = <24>;
+		sample_size = <3>;
 		status = "okay";
 	};
 

samples/caf_sensor_manager/boards/qemu_cortex_m3.overlay

@@ -16,7 +16,7 @@
 		compatible = "caf,aggregator";
 		sensor_descr = "accel_sim_xyz";
 		buf_data_length = <240>;
-		sensor_data_size = <24>;
+		sample_size = <3>;
 		status = "okay";
 	};
 };

samples/caf_sensor_manager/remote/boards/nrf5340dk_nrf5340_cpunet.overlay

@@ -22,7 +22,7 @@
 		compatible = "caf,aggregator";
 		sensor_descr = "accel_sim_xyz";
 		buf_data_length = <240>;
-		sensor_data_size = <24>;
+		sample_size = <3>;
 		memory-region = <&sram0_aggregator_area0>;
 		status = "okay";
 	};

samples/caf_sensor_manager/src/modules/workload_sim.c

@@ -20,9 +20,10 @@ BUILD_ASSERT(WORKLOAD_SIMULATED_THREAD_PRIORITY >= 0);
 LOG_MODULE_REGISTER(MODULE);
 
 struct workload {
-	uint8_t *buf;
+	struct sensor_value *samples;
 	struct k_work work;
 	uint8_t sample_cnt;
+	uint8_t values_in_sample;
 	atomic_t busy;
 	const char *sensor_descr;
 };
@@ -41,8 +42,7 @@ static void simulated_work_handler(struct k_work *work)
 	k_busy_wait(BUSY_TIME * workload->sample_cnt);
 	struct sensor_data_aggregator_release_buffer_event *release_evt =
 		new_sensor_data_aggregator_release_buffer_event();
-
-	release_evt->buf = workload->buf;
+	release_evt->samples = workload->samples;
 	release_evt->sensor_descr = workload->sensor_descr;
 	atomic_clear_bit(&workload->busy, 0);
 	APP_EVENT_SUBMIT(release_evt);
@@ -81,8 +81,9 @@ static bool event_handler(const struct app_event_header *aeh)
 		struct workload *workload = get_free_workload();
 
 		__ASSERT_NO_MSG(workload);
-		workload->buf = event->buf;
+		workload->samples = event->samples;
 		workload->sample_cnt = event->sample_cnt;
+		workload->values_in_sample = event->values_in_sample;
 		workload->sensor_descr = event->sensor_descr;
 		k_work_submit_to_queue(&workload_simulated_work_q, &workload->work);
 

subsys/caf/modules/sensor_data_aggregator.c

@@ -20,46 +20,48 @@ LOG_MODULE_REGISTER(MODULE, CONFIG_CAF_SENSOR_DATA_AGGREGATOR_LOG_LEVEL);
 
 #define DT_DRV_COMPAT caf_aggregator
 
-#define __DEFINE_DATA(i, agg_id, size)							\
+#define __DEFINE_DATA(i, agg_id, size)								\
 	static uint8_t agg_ ## agg_id ## _buff_ ## i ## _data[size] __aligned(4)
 
 #if (DT_INST_NODE_HAS_PROP(agg_id, memory_region))
 
-#define __INITIALIZE_BUF(i, agg_id)							\
-	{										\
-		(uint8_t *) DT_REG_ADDR(DT_INST_PHANDLE(agg_id, memory_region)) +	\
-		i * DT_INST_PROP(agg_id, buf_data_length)				\
+#define __INITIALIZE_BUF(i, agg_id)								\
+	{											\
+		(struct sensor_value *) DT_REG_ADDR(DT_INST_PHANDLE(agg_id, memory_region)) +	\
+		i * (DT_INST_PROP(agg_id, buf_data_length))					\
 	},
 
 
 
-/* buf_len has to be equal to n*sensor_data_size. */
+/* buf_len has to be equal to i*sample_size*sizeof(struct sensor_value). */
 #define __DEFINE_BUF_DATA(i)									\
 	static struct aggregator_buffer agg_ ## i ## _bufs[] = {				\
 		UTIL_LISTIFY(DT_INST_PROP(i, buf_count), __INITIALIZE_BUF, i)			\
 	};											\
-	BUILD_ASSERT((DT_INST_PROP(i, buf_data_length) % DT_INST_PROP(i, sensor_data_size)) == 0,\
+	BUILD_ASSERT((DT_INST_PROP(i, buf_data_length) %					\
+		(DT_INST_PROP(i, sample_size) * sizeof(struct sensor_value))) == 0,		\
 		"Wrong sensor data or buffer size");
 
 #else
-#define __INITIALIZE_BUF(i, agg_id)							\
-	{(uint8_t *) &agg_ ## agg_id ## _buff_ ## i ## _data}
+#define __INITIALIZE_BUF(i, agg_id)								\
+	{(struct sensor_value *) &agg_ ## agg_id ## _buff_ ## i ## _data}
 
-/* buf_len has to be equal to n*sensor_data_size. */
+/* buf_len has to be equal to i*sample_size*sizeof(struct sensor_value). */
 #define __DEFINE_BUF_DATA(i)									\
 	LISTIFY(DT_INST_PROP(i, buf_count), __DEFINE_DATA, (;), i,				\
 		DT_INST_PROP(i, buf_data_length));						\
 	static struct aggregator_buffer agg_ ## i ## _bufs[] = {				\
 		LISTIFY(DT_INST_PROP(i, buf_count), __INITIALIZE_BUF, (,), i)			\
 	};											\
-	BUILD_ASSERT((DT_INST_PROP(i, buf_data_length) % DT_INST_PROP(i, sensor_data_size)) == 0,\
+	BUILD_ASSERT((DT_INST_PROP(i, buf_data_length) %					\
+		(DT_INST_PROP(i, sample_size) * sizeof(struct sensor_value))) == 0,		\
 		"Wrong sensor data or buffer size");
 
 #endif
 
 #define __DEFINE_AGGREGATOR(i)								\
 	[i].sensor_descr = DT_INST_PROP(i, sensor_descr),				\
-	[i].sensor_data_size = DT_INST_PROP(i, sensor_data_size),			\
+	[i].values_in_sample = DT_INST_PROP(i, sample_size),				\
 	[i].buf_count = DT_INST_PROP(i, buf_count),					\
 	[i].buf_len = DT_INST_PROP(i, buf_data_length),					\
 	[i].agg_buffers = (struct aggregator_buffer *) &agg_ ## i ## _bufs,		\
@@ -71,17 +73,17 @@ LOG_MODULE_REGISTER(MODULE, CONFIG_CAF_SENSOR_DATA_AGGREGATOR_LOG_LEVEL);
 	}
 
 struct aggregator_buffer {
-	uint8_t *data;		/* Dynamic data. */
-	bool busy;		/* Buffer status. */
-	uint8_t sample_cnt;	/* Number of samples already saved in the buffer. */
+	struct sensor_value *samples;	/* Dynamic data. */
+	bool busy;			/* Buffer status. */
+	uint8_t sample_cnt;		/* Number of samples already saved in the buffer. */
 };
 
 struct aggregator {
 	const char *sensor_descr;		/* sensor_description of the sensor. */
 	struct aggregator_buffer *agg_buffers;	/* Buffers. */
 	struct aggregator_buffer *active_buf;	/* Active buffer to which data will be placed. */
 	enum sensor_state sensor_state;		/* Sensors state. */
-	const uint8_t sensor_data_size;		/* Size of sensor data in bytes. */
+	const uint8_t values_in_sample;		/* Number of sensor values in a sample. */
 	const uint8_t buf_count;		/* Number of buffers. */
 	const uint8_t buf_len;			/* Size of buffor data in bytes. */
 };
@@ -128,8 +130,8 @@ static void send_buffer(struct aggregator *agg, struct aggregator_buffer *ab)
 {
 	ab->busy = true;
 	struct sensor_data_aggregator_event *event = new_sensor_data_aggregator_event();
-
-	event->buf = ab->data;
+	event->values_in_sample = agg->values_in_sample;
+	event->samples = ab->samples;
 	event->sample_cnt = ab->sample_cnt;
 	event->sensor_state = agg->sensor_state;
 	event->sensor_descr = agg->sensor_descr;
@@ -138,26 +140,28 @@ static void send_buffer(struct aggregator *agg, struct aggregator_buffer *ab)
 
 static int enqueue_sample(struct aggregator *agg, struct sensor_event *event)
 {
-	if (event->dyndata.size != agg->sensor_data_size) {
+	size_t chunk_bytes = agg->values_in_sample * sizeof(struct sensor_value);
+
+	if ((event->dyndata.size) != chunk_bytes) {
 		return -EBADMSG;
 	}
 	if (!agg->active_buf) {
 		return -ENOMEM;
 	}
-	struct aggregator_buffer *ab = agg->active_buf;
 
-	size_t pos = ab->sample_cnt * agg->sensor_data_size;
-	size_t avail = agg->buf_len - pos;
+	struct aggregator_buffer *ab = agg->active_buf;
+	size_t pos_values = ab->sample_cnt * agg->values_in_sample;
+	size_t avail_bytes = agg->buf_len - pos_values * sizeof(struct sensor_value);
 
-	if (avail < agg->sensor_data_size) {
+	if (avail_bytes < chunk_bytes) {
 		__ASSERT_NO_MSG(false);
 		return -ENOMEM;
 	}
-	memcpy(&ab->data[pos], event->dyndata.data, event->dyndata.size);
+	memcpy(&ab->samples[pos_values], (uint8_t *)event->dyndata.data, chunk_bytes);
 	ab->sample_cnt++;
-	avail -= event->dyndata.size;
+	avail_bytes -= chunk_bytes;
 
-	if (avail < agg->sensor_data_size) {
+	if (avail_bytes < chunk_bytes) {
 		send_buffer(agg, ab);
 		agg->active_buf = get_free_buffer(agg);
 	}
@@ -192,7 +196,7 @@ static bool event_handler(const struct app_event_header *aeh)
 		__ASSERT_NO_MSG(agg);
 
 		for (size_t i = 0; i < agg->buf_count; i++) {
-			if (agg->agg_buffers[i].data == event->buf) {
+			if (agg->agg_buffers[i].samples == event->samples) {
 				release_buffer(agg, &agg->agg_buffers[i]);
 				break;
 			}

tests/subsys/caf/sensor_data_aggregator/app.overlay

@@ -9,23 +9,23 @@
 		compatible = "caf,aggregator";
 		sensor_descr = "void_basic_test_sensor";
 		buf_data_length = <160>;
-		sensor_data_size = <16>;
+		sample_size = <2>;
 		status = "okay";
 	};
 
 	agg1: agg1 {
 		compatible = "caf,aggregator";
 		sensor_descr = "void_order_test_sensor";
 		buf_data_length = <80>;
-		sensor_data_size = <8>;
+		sample_size = <1>;
 		status = "okay";
 	};
 
 	agg2: agg2 {
 		compatible = "caf,aggregator";
 		sensor_descr = "void_status_test_sensor";
 		buf_data_length = <80>;
-		sensor_data_size = <8>;
+		sample_size = <1>;
 		status = "okay";
 	};
 };

tests/subsys/caf/sensor_data_aggregator/src/modules/test_data_receiver.c

@@ -37,7 +37,7 @@ static bool app_event_handler(const struct app_event_header *aeh)
 		struct sensor_data_aggregator_release_buffer_event *release_evt =
 		new_sensor_data_aggregator_release_buffer_event();
 
-		release_evt->buf = event->buf;
+		release_evt->samples = event->samples;
 		release_evt->sensor_descr = event->sensor_descr;
 		APP_EVENT_SUBMIT(release_evt);
 
@@ -54,10 +54,11 @@ static bool app_event_handler(const struct app_event_header *aeh)
 		} else if (strcmp(event->sensor_descr, ORDER_TEST_AGG_DESCR) == 0) {
 
 			for (int j = 0; j < SAMPLES_IN_AGG_BUF; j++) {
-				zassert_equal(event->buf[j * sizeof(struct sensor_value) *
-							ORDER_TEST_SENSOR_SAMPLE_SIZE],
-					      order_event_indicator,
-					      "Incorrent event order");
+				uint8_t *event_data = (uint8_t *)&event->samples[j *
+						ORDER_TEST_SENSOR_SAMPLE_SIZE];
+
+				zassert_equal(*event_data, order_event_indicator,
+						"Incorrent event order");
 				order_event_indicator--;
 			}
 
@@ -73,10 +74,10 @@ static bool app_event_handler(const struct app_event_header *aeh)
 		} else if (strcmp(event->sensor_descr, STATUS_TEST_AGG_DESCR) == 0) {
 
 			for (int k = 0; k < STATUS_TEST_SENSOR_EVENTS; k++) {
-				zassert_equal(event->buf[k * sizeof(struct sensor_value) *
-							STATUS_TEST_SENSOR_SAMPLE_SIZE],
-					      k,
-					      "Incorrent event order");
+				uint8_t *event_data = (uint8_t *)&event->samples[k *
+						STATUS_TEST_SENSOR_SAMPLE_SIZE];
+
+				zassert_equal(*event_data, k, "Incorrent event order");
 			}
 
 			struct test_end_event *te = new_test_end_event();