From 552c078c3cf3728620b788e7d6c887450e956c35 Mon Sep 17 00:00:00 2001
From: Yuval Peress <peress@google.com>
Date: Mon, 5 Jun 2023 20:26:19 -0600
Subject: [PATCH 1/3] sensors: Add streaming APIs

Introduce a streaming API that uses the same data path as the async API.

This includes features to the decoder:
* Checking if triggers are present

Adding streaming features built ontop of existing triggers:
* Adding 3 operations to be done on a trigger
  * include - include the data with the trigger information
  * nop - do nothing
  * drop - drop the data (flush)
* Add a new sensor_stream() API to mirror sensor_read() but add an
optional handler to be able to cancel the stream.

Signed-off-by: Yuval Peress <peress@google.com>
topic#sensor_stream
---
 drivers/sensor/CMakeLists.txt                 |   1 +
 drivers/sensor/Kconfig                        |  17 +
 drivers/sensor/default_rtio_sensor.c          |   6 +-
 drivers/sensor/sensor_shell.c                 | 317 +++++++++++++++---
 drivers/sensor/sensor_shell.h                 |  26 ++
 drivers/sensor/sensor_shell_stream.c          | 124 +++++++
 include/zephyr/drivers/sensor.h               | 111 +++++-
 .../sensor_shell/boards/tdk_robokit1.conf     |   3 +
 samples/sensor/sensor_shell/sample.yaml       |   2 +
 9 files changed, 543 insertions(+), 64 deletions(-)
 create mode 100644 drivers/sensor/sensor_shell.h
 create mode 100644 drivers/sensor/sensor_shell_stream.c
 create mode 100644 samples/sensor/sensor_shell/boards/tdk_robokit1.conf

diff --git a/drivers/sensor/CMakeLists.txt b/drivers/sensor/CMakeLists.txt
index 782115ca130f06..4fdfb9821f8a02 100644
--- a/drivers/sensor/CMakeLists.txt
+++ b/drivers/sensor/CMakeLists.txt
@@ -161,5 +161,6 @@ zephyr_library_property(ALLOW_EMPTY TRUE)
 
 zephyr_library_sources_ifdef(CONFIG_USERSPACE sensor_handlers.c)
 zephyr_library_sources_ifdef(CONFIG_SENSOR_SHELL sensor_shell.c)
+zephyr_library_sources_ifdef(CONFIG_SENSOR_SHELL_STREAM sensor_shell_stream.c)
 zephyr_library_sources_ifdef(CONFIG_SENSOR_SHELL_BATTERY shell_battery.c)
 zephyr_library_sources_ifdef(CONFIG_SENSOR_ASYNC_API sensor_decoders_init.c default_rtio_sensor.c)
diff --git a/drivers/sensor/Kconfig b/drivers/sensor/Kconfig
index f406b6d1ffe34e..b34f945fef183f 100644
--- a/drivers/sensor/Kconfig
+++ b/drivers/sensor/Kconfig
@@ -36,6 +36,23 @@ config SENSOR_SHELL
 	help
 	  This shell provides access to basic sensor data.
 
+config SENSOR_SHELL_STREAM
+	bool "Sensor shell 'stream' command"
+	depends on SENSOR_SHELL
+	help
+	  Add the 'stream' subcommand to the sensor shell. When run on drivers that
+	  support streaming (usually hardware FIFO backed), the shell will continue
+	  to print new values as they come until the stream is closed.
+
+config SENSOR_SHELL_THREAD_STACK_SIZE
+	int "Stack size for the sensor shell data processing thread"
+	depends on SENSOR_SHELL_STREAM
+	default 1024
+	help
+	  The sensor shell uses a dedicated thread to process data coming from the
+	  sensors in either one-shot or streaming mode. Use this config to control
+	  the size of that thread's stack.
+
 config SENSOR_SHELL_BATTERY
 	bool "Sensor shell 'battery' command"
 	depends on SHELL
diff --git a/drivers/sensor/default_rtio_sensor.c b/drivers/sensor/default_rtio_sensor.c
index e088b4f61c2889..3f9377c33b3591 100644
--- a/drivers/sensor/default_rtio_sensor.c
+++ b/drivers/sensor/default_rtio_sensor.c
@@ -28,8 +28,10 @@ static void sensor_iodev_submit(struct rtio_iodev_sqe *iodev_sqe)
 
 	if (api->submit != NULL) {
 		api->submit(dev, iodev_sqe);
-	} else {
+	} else if (!cfg->is_streaming) {
 		sensor_submit_fallback(dev, iodev_sqe);
+	} else {
+		rtio_iodev_sqe_err(iodev_sqe, -ENOTSUP);
 	}
 }
 
@@ -235,7 +237,7 @@ static void sensor_submit_fallback(const struct device *dev, struct rtio_iodev_s
 		}
 		sample_idx += num_samples;
 	}
-	LOG_DBG("Total channels in header: %u", header->num_channels);
+	LOG_DBG("Total channels in header: %" PRIu32, header->num_channels);
 	rtio_iodev_sqe_ok(iodev_sqe, 0);
 }
 
diff --git a/drivers/sensor/sensor_shell.c b/drivers/sensor/sensor_shell.c
index a85eab23e43ec3..5d41663c491db8 100644
--- a/drivers/sensor/sensor_shell.c
+++ b/drivers/sensor/sensor_shell.c
@@ -16,6 +16,8 @@
 #include <zephyr/sys/iterable_sections.h>
 #include <zephyr/sys/util.h>
 
+#include "sensor_shell.h"
+
 LOG_MODULE_REGISTER(sensor_shell);
 
 #define SENSOR_GET_HELP                                                                            \
@@ -23,6 +25,11 @@ LOG_MODULE_REGISTER(sensor_shell);
 	"when no channels are provided. Syntax:\n"                                                 \
 	"<device_name> <channel name 0> .. <channel name N>"
 
+#define SENSOR_STREAM_HELP                                                                         \
+	"Start/stop streaming sensor data. Data ready trigger will be used if no triggers "        \
+	"are provided. Syntax:\n"                                                                  \
+	"<device_name> on|off <trigger name> incl|drop|nop"
+
 #define SENSOR_ATTR_GET_HELP                                                                       \
 	"Get the sensor's channel attribute. Syntax:\n"                                            \
 	"<device_name> [<channel_name 0> <attribute_name 0> .. "                                   \
@@ -38,7 +45,7 @@ LOG_MODULE_REGISTER(sensor_shell);
 	"Get or set the trigger type on a sensor. Currently only supports `data_ready`.\n"         \
 	"<device_name> <on/off> <trigger_name>"
 
-const char *sensor_channel_name[SENSOR_CHAN_ALL] = {
+const char *sensor_channel_name[SENSOR_CHAN_COMMON_COUNT] = {
 	[SENSOR_CHAN_ACCEL_X] = "accel_x",
 	[SENSOR_CHAN_ACCEL_Y] = "accel_y",
 	[SENSOR_CHAN_ACCEL_Z] = "accel_z",
@@ -96,6 +103,7 @@ const char *sensor_channel_name[SENSOR_CHAN_ALL] = {
 	[SENSOR_CHAN_GAUGE_DESIGN_VOLTAGE] = "gauge_design_voltage",
 	[SENSOR_CHAN_GAUGE_DESIRED_VOLTAGE] = "gauge_desired_voltage",
 	[SENSOR_CHAN_GAUGE_DESIRED_CHARGING_CURRENT] = "gauge_desired_charging_current",
+	[SENSOR_CHAN_ALL] = "all",
 };
 
 static const char *sensor_attribute_name[SENSOR_ATTR_COMMON_COUNT] = {
@@ -114,6 +122,7 @@ static const char *sensor_attribute_name[SENSOR_ATTR_COMMON_COUNT] = {
 	[SENSOR_ATTR_FEATURE_MASK] = "feature_mask",
 	[SENSOR_ATTR_ALERT] = "alert",
 	[SENSOR_ATTR_FF_DUR] = "ff_dur",
+	[SENSOR_ATTR_FIFO_WATERMARK] = "fifo_wm",
 };
 
 /* Forward declaration */
@@ -146,12 +155,32 @@ static const struct {
 	TRIGGER_DATA_ENTRY(SENSOR_TRIG_FREEFALL, freefall, NULL),
 	TRIGGER_DATA_ENTRY(SENSOR_TRIG_MOTION, motion, NULL),
 	TRIGGER_DATA_ENTRY(SENSOR_TRIG_STATIONARY, stationary, NULL),
+	TRIGGER_DATA_ENTRY(SENSOR_TRIG_FIFO_WATERMARK, fifo_wm, NULL),
+	TRIGGER_DATA_ENTRY(SENSOR_TRIG_FIFO_FULL, fifo_full, NULL),
 };
 
+/**
+ * Lookup the sensor trigger data by name
+ *
+ * @param name The name of the trigger
+ * @return < 0 on error
+ * @return >= 0 if found
+ */
+static int sensor_trigger_name_lookup(const char *name)
+{
+	for (int i = 0; i < ARRAY_SIZE(sensor_trigger_table); ++i) {
+		if (strcmp(name, sensor_trigger_table[i].name) == 0) {
+			return i;
+		}
+	}
+	return -1;
+}
+
 enum dynamic_command_context {
 	NONE,
 	CTX_GET,
 	CTX_ATTR_GET_SET,
+	CTX_STREAM_ON_OFF,
 };
 
 static enum dynamic_command_context current_cmd_ctx = NONE;
@@ -163,6 +192,7 @@ K_MUTEX_DEFINE(cmd_get_mutex);
 static enum sensor_channel iodev_sensor_shell_channels[SENSOR_CHAN_ALL];
 static struct sensor_read_config iodev_sensor_shell_read_config = {
 	.sensor = NULL,
+	.is_streaming = false,
 	.channels = iodev_sensor_shell_channels,
 	.count = 0,
 	.max = ARRAY_SIZE(iodev_sensor_shell_channels),
@@ -234,17 +264,18 @@ static int parse_sensor_value(const char *val_str, struct sensor_value *out)
 	return 0;
 }
 
-struct sensor_shell_processing_context {
-	const struct device *dev;
-	const struct shell *sh;
-};
-
-static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t buf_len,
-					     void *userdata)
+void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t buf_len, void *userdata)
 {
 	struct sensor_shell_processing_context *ctx = userdata;
 	const struct sensor_decoder_api *decoder;
 	uint8_t decoded_buffer[128];
+	struct {
+		uint64_t base_timestamp_ns;
+		int count;
+		uint64_t timestamp_delta;
+		int64_t values[3];
+		int8_t shift;
+	} accumulator_buffer;
 	int rc;
 
 	ARG_UNUSED(buf_len);
@@ -260,6 +291,17 @@ static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t
 		return;
 	}
 
+	for (int trigger = 0; decoder->has_trigger != NULL && trigger < SENSOR_TRIG_COMMON_COUNT;
+	     ++trigger) {
+		if (!decoder->has_trigger(buf, trigger)) {
+			continue;
+		}
+		shell_info(ctx->sh, "Trigger (%d / %s) detected", trigger,
+			   (sensor_trigger_table[trigger].name == NULL
+				    ? "UNKNOWN"
+				    : sensor_trigger_table[trigger].name));
+	}
+
 	for (int channel = 0; channel < SENSOR_CHAN_ALL; ++channel) {
 		uint32_t fit = 0;
 		size_t base_size;
@@ -292,6 +334,7 @@ static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t
 
 		while (decoder->get_frame_count(buf, channel, channel_idx, &frame_count) == 0) {
 			fit = 0;
+			memset(&accumulator_buffer, 0, sizeof(accumulator_buffer));
 			while (decoder->decode(buf, channel, channel_idx, &fit, 1, decoded_buffer) >
 			       0) {
 
@@ -303,41 +346,127 @@ static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t
 					struct sensor_three_axis_data *data =
 						(struct sensor_three_axis_data *)decoded_buffer;
 
-					shell_info(ctx->sh,
-						   "channel idx=%d %s shift=%d "
-						   "value=%" PRIsensor_three_axis_data,
-						   channel, sensor_channel_name[channel],
-						   data->shift,
-						   PRIsensor_three_axis_data_arg(*data, 0));
+					if (accumulator_buffer.count == 0) {
+						accumulator_buffer.base_timestamp_ns =
+							data->header.base_timestamp_ns;
+					}
+					accumulator_buffer.count++;
+					accumulator_buffer.shift = data->shift;
+					accumulator_buffer.timestamp_delta +=
+						data->readings[0].timestamp_delta;
+					accumulator_buffer.values[0] += data->readings[0].values[0];
+					accumulator_buffer.values[1] += data->readings[0].values[1];
+					accumulator_buffer.values[2] += data->readings[0].values[2];
 					break;
 				}
 				case SENSOR_CHAN_PROX: {
 					struct sensor_byte_data *data =
 						(struct sensor_byte_data *)decoded_buffer;
 
-					shell_info(ctx->sh,
-						   "channel idx=%d %s value=%" PRIsensor_byte_data(
-							   is_near),
-						   channel, sensor_channel_name[channel],
-						   PRIsensor_byte_data_arg(*data, 0, is_near));
+					if (accumulator_buffer.count == 0) {
+						accumulator_buffer.base_timestamp_ns =
+							data->header.base_timestamp_ns;
+					}
+					accumulator_buffer.count++;
+					accumulator_buffer.timestamp_delta +=
+						data->readings[0].timestamp_delta;
+					accumulator_buffer.values[0] += data->readings[0].is_near;
 					break;
 				}
 				default: {
 					struct sensor_q31_data *data =
 						(struct sensor_q31_data *)decoded_buffer;
 
-					shell_info(ctx->sh,
-						   "channel idx=%d %s shift=%d "
-						   "value=%" PRIsensor_q31_data,
-						   channel,
-						   (channel >= ARRAY_SIZE(sensor_channel_name))
-							   ? ""
-							   : sensor_channel_name[channel],
-						   data->shift, PRIsensor_q31_data_arg(*data, 0));
+					if (accumulator_buffer.count == 0) {
+						accumulator_buffer.base_timestamp_ns =
+							data->header.base_timestamp_ns;
+					}
+					accumulator_buffer.count++;
+					accumulator_buffer.shift = data->shift;
+					accumulator_buffer.timestamp_delta +=
+						data->readings[0].timestamp_delta;
+					accumulator_buffer.values[0] += data->readings[0].value;
 					break;
 				}
 				}
 			}
+
+			/* Print the accumulated value average */
+			switch (channel) {
+			case SENSOR_CHAN_ACCEL_XYZ:
+			case SENSOR_CHAN_GYRO_XYZ:
+			case SENSOR_CHAN_MAGN_XYZ:
+			case SENSOR_CHAN_POS_DX: {
+				struct sensor_three_axis_data *data =
+					(struct sensor_three_axis_data *)decoded_buffer;
+
+				data->header.base_timestamp_ns =
+					accumulator_buffer.base_timestamp_ns;
+				data->header.reading_count = 1;
+				data->shift = accumulator_buffer.shift;
+				data->readings[0].timestamp_delta =
+					(uint32_t)(accumulator_buffer.timestamp_delta /
+						   accumulator_buffer.count);
+				data->readings[0].values[0] = (q31_t)(accumulator_buffer.values[0] /
+								      accumulator_buffer.count);
+				data->readings[0].values[1] = (q31_t)(accumulator_buffer.values[1] /
+								      accumulator_buffer.count);
+				data->readings[0].values[2] = (q31_t)(accumulator_buffer.values[2] /
+								      accumulator_buffer.count);
+				shell_info(ctx->sh,
+					   "channel idx=%d %s shift=%d num_samples=%d "
+					   "value=%" PRIsensor_three_axis_data,
+					   channel, sensor_channel_name[channel],
+					   data->shift, accumulator_buffer.count,
+					   PRIsensor_three_axis_data_arg(*data, 0));
+				break;
+			}
+			case SENSOR_CHAN_PROX: {
+				struct sensor_byte_data *data =
+					(struct sensor_byte_data *)decoded_buffer;
+
+				data->header.base_timestamp_ns =
+					accumulator_buffer.base_timestamp_ns;
+				data->header.reading_count = 1;
+				data->readings[0].timestamp_delta =
+					(uint32_t)(accumulator_buffer.timestamp_delta /
+						   accumulator_buffer.count);
+				data->readings[0].is_near =
+					accumulator_buffer.values[0] / accumulator_buffer.count;
+
+				shell_info(ctx->sh,
+					   "channel idx=%d %s num_samples=%d "
+					   "value=%" PRIsensor_byte_data(is_near),
+					   channel, sensor_channel_name[channel],
+					   accumulator_buffer.count,
+					   PRIsensor_byte_data_arg(*data, 0, is_near));
+				break;
+			}
+			default: {
+				struct sensor_q31_data *data =
+					(struct sensor_q31_data *)decoded_buffer;
+
+				data->header.base_timestamp_ns =
+					accumulator_buffer.base_timestamp_ns;
+				data->header.reading_count = 1;
+				data->shift = accumulator_buffer.shift;
+				data->readings[0].timestamp_delta =
+					(uint32_t)(accumulator_buffer.timestamp_delta /
+						   accumulator_buffer.count);
+				data->readings[0].value = (q31_t)(accumulator_buffer.values[0] /
+								  accumulator_buffer.count);
+
+				shell_info(ctx->sh,
+					   "channel idx=%d %s shift=%d num_samples=%d "
+					   "value=%" PRIsensor_q31_data,
+					   channel,
+					   (channel >= ARRAY_SIZE(sensor_channel_name))
+						   ? ""
+						   : sensor_channel_name[channel],
+					   data->shift, accumulator_buffer.count,
+					   PRIsensor_q31_data_arg(*data, 0));
+			}
+			}
 			++channel_idx;
 		}
 	}
@@ -345,6 +474,7 @@ static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t
 
 static int cmd_get_sensor(const struct shell *sh, size_t argc, char *argv[])
 {
+	static struct sensor_shell_processing_context ctx;
 	const struct device *dev;
 	int count = 0;
 	int err;
@@ -392,21 +522,27 @@ static int cmd_get_sensor(const struct shell *sh, size_t argc, char *argv[])
 	iodev_sensor_shell_read_config.sensor = dev;
 	iodev_sensor_shell_read_config.count = count;
 
-	struct sensor_shell_processing_context ctx = {
-		.dev = dev,
-		.sh = sh,
-	};
+	ctx.dev = dev;
+	ctx.sh = sh;
 	err = sensor_read(&iodev_sensor_shell_read, &sensor_read_rtio, &ctx);
 	if (err < 0) {
 		shell_error(sh, "Failed to read sensor: %d", err);
 	}
-	sensor_processing_with_callback(&sensor_read_rtio, sensor_shell_processing_callback);
+	if (!IS_ENABLED(CONFIG_SENSOR_SHELL_STREAM)) {
+		/*
+		 * Streaming enables a thread that polls the RTIO context, so if it's enabled, we
+		 * don't need a blocking read here.
+		 */
+		sensor_processing_with_callback(&sensor_read_rtio,
+						sensor_shell_processing_callback);
+	}
 
 	k_mutex_unlock(&cmd_get_mutex);
 
 	return 0;
 }
 
+
 static int cmd_sensor_attr_set(const struct shell *shell_ptr, size_t argc, char *argv[])
 {
 	const struct device *dev;
@@ -517,9 +653,37 @@ static int cmd_sensor_attr_get(const struct shell *shell_ptr, size_t argc, char
 }
 
 static void channel_name_get(size_t idx, struct shell_static_entry *entry);
-
 SHELL_DYNAMIC_CMD_CREATE(dsub_channel_name, channel_name_get);
 
+static void attribute_name_get(size_t idx, struct shell_static_entry *entry);
+SHELL_DYNAMIC_CMD_CREATE(dsub_attribute_name, attribute_name_get);
+
+static void channel_name_get(size_t idx, struct shell_static_entry *entry)
+{
+	int cnt = 0;
+
+	entry->syntax = NULL;
+	entry->handler = NULL;
+	entry->help = NULL;
+	if (current_cmd_ctx == CTX_GET) {
+		entry->subcmd = &dsub_channel_name;
+	} else if (current_cmd_ctx == CTX_ATTR_GET_SET) {
+		entry->subcmd = &dsub_attribute_name;
+	} else {
+		entry->subcmd = NULL;
+	}
+
+	for (int i = 0; i < ARRAY_SIZE(sensor_channel_name); i++) {
+		if (sensor_channel_name[i] != NULL) {
+			if (cnt == idx) {
+				entry->syntax = sensor_channel_name[i];
+				break;
+			}
+			cnt++;
+		}
+	}
+}
+
 static void attribute_name_get(size_t idx, struct shell_static_entry *entry)
 {
 	int cnt = 0;
@@ -529,7 +693,7 @@ static void attribute_name_get(size_t idx, struct shell_static_entry *entry)
 	entry->help = NULL;
 	entry->subcmd = &dsub_channel_name;
 
-	for (int i = 0; i < SENSOR_ATTR_COMMON_COUNT; i++) {
+	for (int i = 0; i < ARRAY_SIZE(sensor_attribute_name); i++) {
 		if (sensor_attribute_name[i] != NULL) {
 			if (cnt == idx) {
 				entry->syntax = sensor_attribute_name[i];
@@ -539,27 +703,46 @@ static void attribute_name_get(size_t idx, struct shell_static_entry *entry)
 		}
 	}
 }
-SHELL_DYNAMIC_CMD_CREATE(dsub_attribute_name, attribute_name_get);
 
-static void channel_name_get(size_t idx, struct shell_static_entry *entry)
+static void trigger_opt_get_for_stream(size_t idx, struct shell_static_entry *entry);
+SHELL_DYNAMIC_CMD_CREATE(dsub_trigger_opt_get_for_stream, trigger_opt_get_for_stream);
+
+static void trigger_opt_get_for_stream(size_t idx, struct shell_static_entry *entry)
+{
+	entry->syntax = NULL;
+	entry->handler = NULL;
+	entry->help = NULL;
+	entry->subcmd = NULL;
+
+	switch (idx) {
+	case SENSOR_STREAM_DATA_INCLUDE:
+		entry->syntax = "incl";
+		break;
+	case SENSOR_STREAM_DATA_DROP:
+		entry->syntax = "drop";
+		break;
+	case SENSOR_STREAM_DATA_NOP:
+		entry->syntax = "nop";
+		break;
+	}
+}
+
+static void trigger_name_get_for_stream(size_t idx, struct shell_static_entry *entry);
+SHELL_DYNAMIC_CMD_CREATE(dsub_trigger_name_for_stream, trigger_name_get_for_stream);
+
+static void trigger_name_get_for_stream(size_t idx, struct shell_static_entry *entry)
 {
 	int cnt = 0;
 
 	entry->syntax = NULL;
 	entry->handler = NULL;
 	entry->help = NULL;
-	if (current_cmd_ctx == CTX_GET) {
-		entry->subcmd = &dsub_channel_name;
-	} else if (current_cmd_ctx == CTX_ATTR_GET_SET) {
-		entry->subcmd = &dsub_attribute_name;
-	} else {
-		entry->subcmd = NULL;
-	}
+	entry->subcmd = &dsub_trigger_opt_get_for_stream;
 
-	for (int i = 0; i < SENSOR_CHAN_ALL; i++) {
-		if (sensor_channel_name[i] != NULL) {
+	for (int i = 0; i < ARRAY_SIZE(sensor_trigger_table); i++) {
+		if (sensor_trigger_table[i].name != NULL) {
 			if (cnt == idx) {
-				entry->syntax = sensor_channel_name[i];
+				entry->syntax = sensor_trigger_table[i].name;
 				break;
 			}
 			cnt++;
@@ -567,6 +750,22 @@ static void channel_name_get(size_t idx, struct shell_static_entry *entry)
 	}
 }
 
+static void stream_on_off(size_t idx, struct shell_static_entry *entry)
+{
+	entry->syntax = NULL;
+	entry->handler = NULL;
+	entry->help = NULL;
+
+	if (idx == 0) {
+		entry->syntax = "on";
+		entry->subcmd = &dsub_trigger_name_for_stream;
+	} else if (idx == 1) {
+		entry->syntax = "off";
+		entry->subcmd = NULL;
+	}
+}
+SHELL_DYNAMIC_CMD_CREATE(dsub_stream_on_off, stream_on_off);
+
 static void device_name_get(size_t idx, struct shell_static_entry *entry);
 
 SHELL_DYNAMIC_CMD_CREATE(dsub_device_name, device_name_get);
@@ -603,7 +802,7 @@ static void trigger_name_get(size_t idx, struct shell_static_entry *entry)
 	entry->help = NULL;
 	entry->subcmd = NULL;
 
-	for (int i = 0; i < SENSOR_TRIG_COMMON_COUNT; i++) {
+	for (int i = 0; i < ARRAY_SIZE(sensor_trigger_table); i++) {
 		if (sensor_trigger_table[i].name != NULL) {
 			if (cnt == idx) {
 				entry->syntax = sensor_trigger_table[i].name;
@@ -649,6 +848,18 @@ static void device_name_get_for_trigger(size_t idx, struct shell_static_entry *e
 
 SHELL_DYNAMIC_CMD_CREATE(dsub_trigger, device_name_get_for_trigger);
 
+static void device_name_get_for_stream(size_t idx, struct shell_static_entry *entry)
+{
+	const struct device *dev = shell_device_lookup(idx, NULL);
+
+	current_cmd_ctx = CTX_STREAM_ON_OFF;
+	entry->syntax = (dev != NULL) ? dev->name : NULL;
+	entry->handler = NULL;
+	entry->help = NULL;
+	entry->subcmd = &dsub_stream_on_off;
+}
+SHELL_DYNAMIC_CMD_CREATE(dsub_device_name_for_stream, device_name_get_for_stream);
+
 static int cmd_get_sensor_info(const struct shell *sh, size_t argc, char **argv)
 {
 	ARG_UNUSED(argc);
@@ -738,7 +949,7 @@ static void data_ready_trigger_handler(const struct device *sensor,
 static int cmd_trig_sensor(const struct shell *sh, size_t argc, char **argv)
 {
 	const struct device *dev;
-	enum sensor_trigger_type trigger;
+	int trigger;
 	int err;
 
 	if (argc < 4) {
@@ -754,12 +965,8 @@ static int cmd_trig_sensor(const struct shell *sh, size_t argc, char **argv)
 	}
 
 	/* Map the trigger string to an enum value */
-	for (trigger = 0; trigger < ARRAY_SIZE(sensor_trigger_table); trigger++) {
-		if (strcmp(argv[3], sensor_trigger_table[trigger].name) == 0) {
-			break;
-		}
-	}
-	if (trigger >= SENSOR_TRIG_COMMON_COUNT || sensor_trigger_table[trigger].handler == NULL) {
+	trigger = sensor_trigger_name_lookup(argv[3]);
+	if (trigger < 0 || sensor_trigger_table[trigger].handler == NULL) {
 		shell_error(sh, "Unsupported trigger type (%s)", argv[3]);
 		return -ENOTSUP;
 	}
@@ -792,6 +999,8 @@ SHELL_STATIC_SUBCMD_SET_CREATE(sub_sensor,
 			cmd_sensor_attr_set, 2, 255),
 	SHELL_CMD_ARG(attr_get, &dsub_device_name_for_attr, SENSOR_ATTR_GET_HELP,
 			cmd_sensor_attr_get, 2, 255),
+	SHELL_COND_CMD(CONFIG_SENSOR_SHELL_STREAM, stream, &dsub_device_name_for_stream,
+			SENSOR_STREAM_HELP, cmd_sensor_stream),
 	SHELL_COND_CMD(CONFIG_SENSOR_INFO, info, NULL, SENSOR_INFO_HELP,
 			cmd_get_sensor_info),
 	SHELL_CMD_ARG(trig, &dsub_trigger, SENSOR_TRIG_HELP, cmd_trig_sensor,
diff --git a/drivers/sensor/sensor_shell.h b/drivers/sensor/sensor_shell.h
new file mode 100644
index 00000000000000..cfc8b2489db3cf
--- /dev/null
+++ b/drivers/sensor/sensor_shell.h
@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2023 Google LLC
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef ZEPHYR_DRIVERS_SENSOR_SENSOR_SHELL_H
+#define ZEPHYR_DRIVERS_SENSOR_SENSOR_SHELL_H
+
+#include <zephyr/device.h>
+#include <zephyr/drivers/sensor.h>
+#include <zephyr/rtio/rtio.h>
+#include <zephyr/shell/shell.h>
+
+struct sensor_shell_processing_context {
+	const struct device *dev;
+	const struct shell *sh;
+};
+
+extern struct rtio sensor_read_rtio;
+
+int cmd_sensor_stream(const struct shell *shell_ptr, size_t argc, char *argv[]);
+
+void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t buf_len, void *userdata);
+
+#endif /* ZEPHYR_DRIVERS_SENSOR_SENSOR_SHELL_H */
diff --git a/drivers/sensor/sensor_shell_stream.c b/drivers/sensor/sensor_shell_stream.c
new file mode 100644
index 00000000000000..ab3b835aa10386
--- /dev/null
+++ b/drivers/sensor/sensor_shell_stream.c
@@ -0,0 +1,124 @@
+/*
+ * Copyright (c) 2023 Google LLC
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <stdlib.h>
+#include <string.h>
+
+#include <zephyr/drivers/sensor.h>
+#include <zephyr/kernel.h>
+
+#include "sensor_shell.h"
+
+/* Create a single common config for streaming */
+static struct sensor_stream_trigger iodev_sensor_shell_trigger;
+static struct sensor_read_config iodev_sensor_shell_stream_config = {
+	.sensor = NULL,
+	.is_streaming = true,
+	.triggers = &iodev_sensor_shell_trigger,
+	.count = 0,
+	.max = 1,
+};
+RTIO_IODEV_DEFINE(iodev_sensor_shell_stream, &__sensor_iodev_api,
+		  &iodev_sensor_shell_stream_config);
+
+static void sensor_shell_processing_entry_point(void *a, void *b, void *c)
+{
+	ARG_UNUSED(a);
+	ARG_UNUSED(b);
+	ARG_UNUSED(c);
+
+	while (true) {
+		sensor_processing_with_callback(&sensor_read_rtio,
+						sensor_shell_processing_callback);
+	}
+}
+K_THREAD_DEFINE(sensor_shell_processing_tid, CONFIG_SENSOR_SHELL_THREAD_STACK_SIZE,
+		sensor_shell_processing_entry_point, NULL, NULL, NULL, 0, 0, 0);
+
+int cmd_sensor_stream(const struct shell *shell_ptr, size_t argc, char *argv[])
+{
+	static struct rtio_sqe *current_streaming_handle;
+	static struct sensor_shell_processing_context ctx;
+	const struct device *dev = device_get_binding(argv[1]);
+
+	if (argc != 5 && argc != 3) {
+		shell_error(shell_ptr, "Wrong number of arguments (%zu)", argc);
+		return -EINVAL;
+	}
+
+	if (dev == NULL) {
+		shell_error(shell_ptr, "Device unknown (%s)", argv[1]);
+		return -ENODEV;
+	}
+
+	if (current_streaming_handle != NULL) {
+		shell_info(shell_ptr, "Disabling existing stream");
+		rtio_sqe_cancel(current_streaming_handle);
+	}
+
+	if (strcmp("off", argv[2]) == 0) {
+		return 0;
+	}
+
+	if (strcmp("on", argv[2]) != 0) {
+		shell_error(shell_ptr, "Unknown streaming operation (%s)", argv[2]);
+		return -EINVAL;
+	}
+
+	if (strcmp("double_tap", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_DOUBLE_TAP;
+	} else if (strcmp("data_ready", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_DATA_READY;
+	} else if (strcmp("delta", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_DELTA;
+	} else if (strcmp("freefall", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_FREEFALL;
+	} else if (strcmp("motion", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_MOTION;
+	} else if (strcmp("near_far", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_NEAR_FAR;
+	} else if (strcmp("stationary", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_STATIONARY;
+	} else if (strcmp("threshold", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_THRESHOLD;
+	} else if (strcmp("fifo_wm", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_FIFO_WATERMARK;
+	} else if (strcmp("fifo_full", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_FIFO_FULL;
+	} else if (strcmp("tap", argv[3]) == 0) {
+		iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_TAP;
+	} else {
+		shell_error(shell_ptr, "Invalid trigger (%s)", argv[3]);
+		return -EINVAL;
+	}
+
+	if (strcmp("incl", argv[4]) == 0) {
+		iodev_sensor_shell_trigger.opt = SENSOR_STREAM_DATA_INCLUDE;
+	} else if (strcmp("drop", argv[4]) == 0) {
+		iodev_sensor_shell_trigger.opt = SENSOR_STREAM_DATA_DROP;
+	} else if (strcmp("nop", argv[4]) == 0) {
+		iodev_sensor_shell_trigger.opt = SENSOR_STREAM_DATA_NOP;
+	} else {
+		shell_error(shell_ptr, "Unknown trigger op (%s)", argv[4]);
+		return -EINVAL;
+	}
+
+	shell_print(shell_ptr, "Enabling stream...");
+	iodev_sensor_shell_stream_config.sensor = dev;
+
+	iodev_sensor_shell_stream_config.count = 1;
+
+	ctx.dev = dev;
+	ctx.sh = shell_ptr;
+
+	int rc = sensor_stream(&iodev_sensor_shell_stream, &sensor_read_rtio, &ctx,
+			       &current_streaming_handle);
+
+	if (rc != 0) {
+		shell_error(shell_ptr, "Failed to start stream");
+	}
+	return rc;
+}
diff --git a/include/zephyr/drivers/sensor.h b/include/zephyr/drivers/sensor.h
index 903acb6f03ac46..715d5ff5385975 100644
--- a/include/zephyr/drivers/sensor.h
+++ b/include/zephyr/drivers/sensor.h
@@ -251,6 +251,12 @@ enum sensor_trigger_type {
 
 	/** Trigger fires when no motion has been detected for a while. */
 	SENSOR_TRIG_STATIONARY,
+
+	/** Trigger fires when the FIFO watermark has been reached. */
+	SENSOR_TRIG_FIFO_WATERMARK,
+
+	/** Trigger fires when the FIFO becomes full. */
+	SENSOR_TRIG_FIFO_FULL,
 	/**
 	 * Number of all common sensor triggers.
 	 */
@@ -328,6 +334,8 @@ enum sensor_attribute {
 	 *  to the new sampling frequency.
 	 */
 	SENSOR_ATTR_FF_DUR,
+	/** Watermark % for the hardware fifo interrupt */
+	SENSOR_ATTR_FIFO_WATERMARK,
 	/**
 	 * Number of all common sensor attributes.
 	 */
@@ -466,6 +474,15 @@ struct sensor_decoder_api {
 	 */
 	int (*decode)(const uint8_t *buffer, enum sensor_channel channel, size_t channel_idx,
 		      uint32_t *fit, uint16_t max_count, void *data_out);
+
+	/**
+	 * @brief Check if the given trigger type is present
+	 *
+	 * @param[in] buffer The buffer provided on the @ref rtio context
+	 * @param[in] trigger The trigger type in question
+	 * @return Whether the trigger is present in the buffer
+	 */
+	bool (*has_trigger)(const uint8_t *buffer, enum sensor_trigger_type trigger);
 };
 
 /**
@@ -538,13 +555,38 @@ int sensor_natively_supported_channel_size_info(enum sensor_channel channel, siz
 typedef int (*sensor_get_decoder_t)(const struct device *dev,
 				    const struct sensor_decoder_api **api);
 
+/**
+ * @brief Options for what to do with the associated data when a trigger is consumed
+ */
+enum sensor_stream_data_opt {
+	/** @brief Include whatever data is associated with the trigger */
+	SENSOR_STREAM_DATA_INCLUDE = 0,
+	/** @brief Do nothing with the associated trigger data, it may be consumed later */
+	SENSOR_STREAM_DATA_NOP = 1,
+	/** @brief Flush/clear whatever data is associated with the trigger */
+	SENSOR_STREAM_DATA_DROP = 2,
+};
+
+struct sensor_stream_trigger {
+	enum sensor_trigger_type trigger;
+	enum sensor_stream_data_opt opt;
+};
+
+#define SENSOR_STREAM_TRIGGER_PREP(_trigger, _opt)                                                 \
+	{                                                                                          \
+		.trigger = (_trigger), .opt = (_opt),                                              \
+	}
 /*
  * Internal data structure used to store information about the IODevice for async reading and
  * streaming sensor data.
  */
 struct sensor_read_config {
 	const struct device *sensor;
-	enum sensor_channel *const channels;
+	const bool is_streaming;
+	union {
+		enum sensor_channel *const channels;
+		struct sensor_stream_trigger *const triggers;
+	};
 	size_t count;
 	const size_t max;
 };
@@ -564,14 +606,45 @@ struct sensor_read_config {
  * @endcode
  */
 #define SENSOR_DT_READ_IODEV(name, dt_node, ...)                                                   \
-	static enum sensor_channel __channel_array_##name[] = {__VA_ARGS__};                       \
-	static struct sensor_read_config __sensor_read_config_##name = {                           \
+	static enum sensor_channel _CONCAT(__channel_array_, name)[] = {__VA_ARGS__};              \
+	static struct sensor_read_config _CONCAT(__sensor_read_config_, name) = {                  \
 		.sensor = DEVICE_DT_GET(dt_node),                                                  \
-		.channels = __channel_array_##name,                                                \
-		.count = ARRAY_SIZE(__channel_array_##name),                                       \
-		.max = ARRAY_SIZE(__channel_array_##name),                                         \
+		.is_streaming = false,                                                             \
+		.channels = _CONCAT(__channel_array_, name),                                       \
+		.count = ARRAY_SIZE(_CONCAT(__channel_array_, name)),                              \
+		.max = ARRAY_SIZE(_CONCAT(__channel_array_, name)),                                \
 	};                                                                                         \
-	RTIO_IODEV_DEFINE(name, &__sensor_iodev_api, &__sensor_read_config_##name)
+	RTIO_IODEV_DEFINE(name, &__sensor_iodev_api, _CONCAT(&__sensor_read_config_, name))
+
+/**
+ * @brief Define a stream instance of a sensor
+ *
+ * Use this macro to generate a @ref rtio_iodev for starting a stream that's triggered by specific
+ * interrupts. Example:
+ *
+ * @code(.c)
+ * SENSOR_DT_STREAM_IODEV(imu_stream, DT_ALIAS(imu),
+ *     {SENSOR_TRIG_FIFO_WATERMARK, SENSOR_STREAM_DATA_INCLUDE},
+ *     {SENSOR_TRIG_FIFO_FULL, SENSOR_STREAM_DATA_NOP});
+ *
+ * int main(void) {
+ *   struct rtio_sqe *handle;
+ *   sensor_stream(&imu_stream, &rtio, NULL, &handle);
+ *   k_msleep(1000);
+ *   rtio_sqe_cancel(handle);
+ * }
+ * @endcode
+ */
+#define SENSOR_DT_STREAM_IODEV(name, dt_node, ...)                                                 \
+	static struct sensor_stream_trigger _CONCAT(__trigger_array_, name)[] = {__VA_ARGS__};     \
+	static struct sensor_read_config _CONCAT(__sensor_read_config_, name) = {                  \
+		.sensor = DEVICE_DT_GET(dt_node),                                                  \
+		.is_streaming = true,                                                              \
+		.triggers = _CONCAT(__trigger_array_, name),                                       \
+		.count = ARRAY_SIZE(_CONCAT(__trigger_array_, name)),                              \
+		.max = ARRAY_SIZE(_CONCAT(__trigger_array_, name)),                                \
+	};                                                                                         \
+	RTIO_IODEV_DEFINE(name, &__sensor_iodev_api, &_CONCAT(__sensor_read_config_, name))
 
 /* Used to submit an RTIO sqe to the sensor's iodev */
 typedef int (*sensor_submit_t)(const struct device *sensor, struct rtio_iodev_sqe *sqe);
@@ -880,7 +953,7 @@ static inline int z_impl_sensor_reconfigure_read_iodev(struct rtio_iodev *iodev,
 {
 	struct sensor_read_config *cfg = (struct sensor_read_config *)iodev->data;
 
-	if (cfg->max < num_channels) {
+	if (cfg->max < num_channels || cfg->is_streaming) {
 		return -ENOMEM;
 	}
 
@@ -888,6 +961,28 @@ static inline int z_impl_sensor_reconfigure_read_iodev(struct rtio_iodev *iodev,
 	memcpy(cfg->channels, channels, num_channels * sizeof(enum sensor_channel));
 	cfg->count = num_channels;
 	return 0;
+}
+
+static inline int sensor_stream(struct rtio_iodev *iodev, struct rtio *ctx, void *userdata,
+				struct rtio_sqe **handle)
+{
+	if (IS_ENABLED(CONFIG_USERSPACE)) {
+		struct rtio_sqe sqe;
+
+		rtio_sqe_prep_read_multishot(&sqe, iodev, RTIO_PRIO_NORM, userdata);
+		rtio_sqe_copy_in_get_handles(ctx, &sqe, handle, 1);
+	} else {
+		struct rtio_sqe *sqe = rtio_sqe_acquire(ctx);
+
+		if (sqe == NULL) {
+			return -ENOMEM;
+		}
+		if (handle != NULL) {
+			*handle = sqe;
+		}
+		rtio_sqe_prep_read_multishot(sqe, iodev, RTIO_PRIO_NORM, userdata);
+	}
+	rtio_submit(ctx, 0);
 	return 0;
 }
 
diff --git a/samples/sensor/sensor_shell/boards/tdk_robokit1.conf b/samples/sensor/sensor_shell/boards/tdk_robokit1.conf
new file mode 100644
index 00000000000000..aee20b8b10049c
--- /dev/null
+++ b/samples/sensor/sensor_shell/boards/tdk_robokit1.conf
@@ -0,0 +1,3 @@
+# Copyright (c) 2023 Google LLC
+# SPDX-License-Identifier: Apache-2.0
+CONFIG_SPI_RTIO=y
diff --git a/samples/sensor/sensor_shell/sample.yaml b/samples/sensor/sensor_shell/sample.yaml
index 7b194f4d622781..d89dfebc21aae2 100644
--- a/samples/sensor/sensor_shell/sample.yaml
+++ b/samples/sensor/sensor_shell/sample.yaml
@@ -4,6 +4,8 @@ tests:
   sample.sensor.shell:
     integration_platforms:
       - frdm_k64f
+    # TODO Remove once #63414 is resolved
+    platform_exclude: gd32l233r_eval
     filter: ( CONFIG_UART_CONSOLE and CONFIG_SERIAL_SUPPORT_INTERRUPT )
     tags: shell
     harness: keyboard

From a4e629406321c0a9ee2de5a1cc9be0b8c0ab91d1 Mon Sep 17 00:00:00 2001
From: Yuval Peress <peress@google.com>
Date: Wed, 5 Jul 2023 13:42:04 -0600
Subject: [PATCH 2/3] icm42688: Implement streaming APIs

Add streaming implementation for icm42688 using both threshold and
full FIFO triggers.

Signed-off-by: Yuval Peress <peress@google.com>
topic#sensor_stream
---
 drivers/sensor/icm42688/CMakeLists.txt        |   1 +
 drivers/sensor/icm42688/Kconfig               |  10 +
 drivers/sensor/icm42688/icm42688.c            |  32 +-
 drivers/sensor/icm42688/icm42688.h            |  12 +
 drivers/sensor/icm42688/icm42688_common.c     |   9 +-
 drivers/sensor/icm42688/icm42688_decoder.c    | 352 +++++++++++++++++-
 drivers/sensor/icm42688/icm42688_decoder.h    |   9 +
 drivers/sensor/icm42688/icm42688_reg.h        |   8 +
 drivers/sensor/icm42688/icm42688_rtio.c       |  16 +-
 drivers/sensor/icm42688/icm42688_rtio.h       |   7 +
 .../sensor/icm42688/icm42688_rtio_stream.c    | 319 ++++++++++++++++
 drivers/sensor/icm42688/icm42688_trigger.c    |  24 +-
 drivers/sensor/icm42688/icm42688_trigger.h    |   3 +-
 13 files changed, 773 insertions(+), 29 deletions(-)
 create mode 100644 drivers/sensor/icm42688/icm42688_rtio_stream.c

diff --git a/drivers/sensor/icm42688/CMakeLists.txt b/drivers/sensor/icm42688/CMakeLists.txt
index cf9308bb95c447..fbc63b6a4b2291 100644
--- a/drivers/sensor/icm42688/CMakeLists.txt
+++ b/drivers/sensor/icm42688/CMakeLists.txt
@@ -10,6 +10,7 @@ zephyr_library_sources(
 
 zephyr_library_sources_ifdef(CONFIG_SENSOR_ASYNC_API icm42688_rtio.c)
 zephyr_library_sources_ifdef(CONFIG_ICM42688_DECODER icm42688_decoder.c)
+zephyr_library_sources_ifdef(CONFIG_ICM42688_STREAM icm42688_rtio_stream.c)
 zephyr_library_sources_ifdef(CONFIG_ICM42688_TRIGGER icm42688_trigger.c)
 zephyr_library_sources_ifdef(CONFIG_EMUL_ICM42688 icm42688_emul.c)
 zephyr_include_directories_ifdef(CONFIG_EMUL_ICM42688 .)
diff --git a/drivers/sensor/icm42688/Kconfig b/drivers/sensor/icm42688/Kconfig
index 2944a9d1183ad4..d7ee43bc7618bf 100644
--- a/drivers/sensor/icm42688/Kconfig
+++ b/drivers/sensor/icm42688/Kconfig
@@ -33,6 +33,7 @@ if ICM42688
 
 choice
 	prompt "Trigger mode"
+	default ICM42688_TRIGGER_NONE if ICM42688_STREAM
 	default ICM42688_TRIGGER_GLOBAL_THREAD
 	help
 	  Specify the type of triggering to be used by the driver
@@ -50,6 +51,15 @@ config ICM42688_TRIGGER_OWN_THREAD
 
 endchoice
 
+config ICM42688_STREAM
+	bool "Use hardware FIFO to stream data"
+	select ICM42688_TRIGGER
+	default y
+	depends on SPI_RTIO
+	depends on SENSOR_ASYNC_API
+	help
+	  Use this config option to enable streaming sensor data via RTIO subsystem.
+
 config ICM42688_TRIGGER
 	bool
 
diff --git a/drivers/sensor/icm42688/icm42688.c b/drivers/sensor/icm42688/icm42688.c
index 43fd6a72f47f68..ee92c30987ecfc 100644
--- a/drivers/sensor/icm42688/icm42688.c
+++ b/drivers/sensor/icm42688/icm42688.c
@@ -82,7 +82,7 @@ int icm42688_channel_parse_readings(enum sensor_channel chan, int16_t readings[7
 }
 
 static int icm42688_channel_get(const struct device *dev, enum sensor_channel chan,
-			 struct sensor_value *val)
+				struct sensor_value *val)
 {
 	struct icm42688_dev_data *data = dev->data;
 
@@ -156,6 +156,19 @@ static int icm42688_attr_set(const struct device *dev, enum sensor_channel chan,
 			res = -EINVAL;
 		}
 		break;
+	case SENSOR_CHAN_ALL:
+		if (attr == SENSOR_ATTR_FIFO_WATERMARK) {
+			int64_t mval = sensor_value_to_micro(val);
+
+			if (mval < 0 || mval > 1000000) {
+				return -EINVAL;
+			}
+			new_config.fifo_wm = CLAMP(mval * 2048 / 1000000, 0, 2048);
+		} else {
+			LOG_ERR("Unsupported attribute");
+			res = -EINVAL;
+		}
+		break;
 	default:
 		LOG_ERR("Unsupported channel");
 		res = -EINVAL;
@@ -257,7 +270,13 @@ int icm42688_init(const struct device *dev)
 	data->cfg.gyro_mode = ICM42688_GYRO_LN;
 	data->cfg.gyro_fs = ICM42688_GYRO_FS_125;
 	data->cfg.gyro_odr = ICM42688_GYRO_ODR_1000;
-	data->cfg.fifo_en = false;
+	data->cfg.temp_dis = false;
+	data->cfg.fifo_en = IS_ENABLED(CONFIG_ICM42688_STREAM);
+	data->cfg.fifo_wm = 0;
+	data->cfg.fifo_hires = 0;
+	data->cfg.interrupt1_drdy = 0;
+	data->cfg.interrupt1_fifo_ths = 0;
+	data->cfg.interrupt1_fifo_full = 0;
 
 	res = icm42688_configure(dev, &data->cfg);
 	if (res != 0) {
@@ -283,8 +302,15 @@ void icm42688_unlock(const struct device *dev)
 #define ICM42688_SPI_CFG                                                                           \
 	SPI_OP_MODE_MASTER | SPI_MODE_CPOL | SPI_MODE_CPHA | SPI_WORD_SET(8) | SPI_TRANSFER_MSB
 
+#define ICM42688_RTIO_DEFINE(inst)                                                                 \
+	SPI_DT_IODEV_DEFINE(icm42688_spi_iodev_##inst, DT_DRV_INST(inst), ICM42688_SPI_CFG, 0U);   \
+	RTIO_DEFINE(icm42688_rtio_##inst, 8, 4);
+
 #define ICM42688_DEFINE_DATA(inst)                                                                 \
-	static struct icm42688_dev_data icm42688_driver_##inst;
+	IF_ENABLED(CONFIG_ICM42688_STREAM, (ICM42688_RTIO_DEFINE(inst)));                          \
+	static struct icm42688_dev_data icm42688_driver_##inst = {                                 \
+		IF_ENABLED(CONFIG_ICM42688_STREAM, (.r = &icm42688_rtio_##inst,                    \
+						    .spi_iodev = &icm42688_spi_iodev_##inst,))};
 
 #define ICM42688_INIT(inst)                                                                        \
 	ICM42688_DEFINE_DATA(inst);                                                                \
diff --git a/drivers/sensor/icm42688/icm42688.h b/drivers/sensor/icm42688/icm42688.h
index 18ae4606e72593..c4167e6a3b9cf4 100644
--- a/drivers/sensor/icm42688/icm42688.h
+++ b/drivers/sensor/icm42688/icm42688.h
@@ -385,6 +385,9 @@ struct icm42688_cfg {
 	/* TODO additional FIFO options */
 
 	/* TODO interrupt options */
+	bool interrupt1_drdy;
+	bool interrupt1_fifo_ths;
+	bool interrupt1_fifo_full;
 };
 
 struct icm42688_trigger_entry {
@@ -405,6 +408,15 @@ struct icm42688_dev_data {
 #elif defined(CONFIG_ICM42688_TRIGGER_GLOBAL_THREAD)
 	struct k_work work;
 #endif
+#ifdef CONFIG_ICM42688_STREAM
+	struct rtio_iodev_sqe *streaming_sqe;
+	struct rtio *r;
+	struct rtio_iodev *spi_iodev;
+	uint8_t int_status;
+	uint16_t fifo_count;
+	uint64_t timestamp;
+	atomic_t reading_fifo;
+#endif /* CONFIG_ICM42688_STREAM */
 	const struct device *dev;
 	struct gpio_callback gpio_cb;
 	sensor_trigger_handler_t data_ready_handler;
diff --git a/drivers/sensor/icm42688/icm42688_common.c b/drivers/sensor/icm42688/icm42688_common.c
index 3ccd107cfbb3ef..9cb3037ed06ea3 100644
--- a/drivers/sensor/icm42688/icm42688_common.c
+++ b/drivers/sensor/icm42688/icm42688_common.c
@@ -12,6 +12,7 @@
 #include "icm42688.h"
 #include "icm42688_reg.h"
 #include "icm42688_spi.h"
+#include "icm42688_trigger.h"
 
 #include <zephyr/logging/log.h>
 LOG_MODULE_REGISTER(ICM42688_LL, CONFIG_SENSOR_LOG_LEVEL);
@@ -165,8 +166,12 @@ int icm42688_configure(const struct device *dev, struct icm42688_cfg *cfg)
 	}
 
 	/* Pulse mode with async reset (resets interrupt line on int status read) */
-	res = icm42688_spi_single_write(&dev_cfg->spi, REG_INT_CONFIG,
-					BIT_INT1_DRIVE_CIRCUIT | BIT_INT1_POLARITY);
+	if (IS_ENABLED(CONFIG_ICM42688_TRIGGER)) {
+		res = icm42688_trigger_enable_interrupt(dev, cfg);
+	} else {
+		res = icm42688_spi_single_write(&dev_cfg->spi, REG_INT_CONFIG,
+						BIT_INT1_DRIVE_CIRCUIT | BIT_INT1_POLARITY);
+	}
 	if (res) {
 		LOG_ERR("Error writing to INT_CONFIG");
 		return res;
diff --git a/drivers/sensor/icm42688/icm42688_decoder.c b/drivers/sensor/icm42688/icm42688_decoder.c
index 0a340aff369567..3dc556ac61c7c3 100644
--- a/drivers/sensor/icm42688/icm42688_decoder.c
+++ b/drivers/sensor/icm42688/icm42688_decoder.c
@@ -198,9 +198,262 @@ int icm42688_encode(const struct device *dev, const enum sensor_channel *const c
 	return 0;
 }
 
+#define IS_ACCEL(chan) ((chan) >= SENSOR_CHAN_ACCEL_X && (chan) <= SENSOR_CHAN_ACCEL_XYZ)
+#define IS_GYRO(chan)  ((chan) >= SENSOR_CHAN_GYRO_X && (chan) <= SENSOR_CHAN_GYRO_XYZ)
+
+static inline q31_t icm42688_read_temperature_from_packet(const uint8_t *pkt)
+{
+	int32_t temperature;
+	int32_t whole;
+	int32_t fraction;
+
+	/* Temperature always assumes a shift of 9 for a range of (-273,273) C */
+	if (FIELD_GET(FIFO_HEADER_20, pkt[0]) == 1) {
+		temperature = (pkt[0xd] << 8) | pkt[0xe];
+
+		icm42688_temp_c(temperature, &whole, &fraction);
+	} else {
+		if (FIELD_GET(FIFO_HEADER_ACCEL, pkt[0]) == 1 &&
+		    FIELD_GET(FIFO_HEADER_GYRO, pkt[0]) == 1) {
+			temperature = pkt[0xd];
+		} else {
+			temperature = pkt[0x7];
+		}
+
+		int64_t sensitivity = 207;
+		int64_t temperature100 = (temperature * 100) + (25 * sensitivity);
+
+		whole = temperature100 / sensitivity;
+		fraction =
+			((temperature100 - whole * sensitivity) * INT64_C(1000000)) / sensitivity;
+	}
+	__ASSERT_NO_MSG(whole >= -512 && whole <= 511);
+	return FIELD_PREP(GENMASK(31, 22), whole) | (fraction * GENMASK64(21, 0) / 1000000);
+}
+
+static int icm42688_read_imu_from_packet(const uint8_t *pkt, bool is_accel, int fs,
+					 uint8_t axis_offset, q31_t *out)
+{
+	int32_t value;
+	int64_t scale = 0;
+	int32_t max = BIT(15);
+	int offset = 1 + (axis_offset * 2);
+
+	if (is_accel) {
+		switch (fs) {
+		case ICM42688_ACCEL_FS_2G:
+			scale = INT64_C(2) * BIT(31 - 5) * 9.80665;
+			break;
+		case ICM42688_ACCEL_FS_4G:
+			scale = INT64_C(4) * BIT(31 - 6) * 9.80665;
+			break;
+		case ICM42688_ACCEL_FS_8G:
+			scale = INT64_C(8) * BIT(31 - 7) * 9.80665;
+			break;
+		case ICM42688_ACCEL_FS_16G:
+			scale = INT64_C(16) * BIT(31 - 8) * 9.80665;
+			break;
+		}
+	} else {
+		switch (fs) {
+		case ICM42688_GYRO_FS_2000:
+			scale = 164;
+			break;
+		case ICM42688_GYRO_FS_1000:
+			scale = 328;
+			break;
+		case ICM42688_GYRO_FS_500:
+			scale = 655;
+			break;
+		case ICM42688_GYRO_FS_250:
+			scale = 1310;
+			break;
+		case ICM42688_GYRO_FS_125:
+			scale = 2620;
+			break;
+		case ICM42688_GYRO_FS_62_5:
+			scale = 5243;
+			break;
+		case ICM42688_GYRO_FS_31_25:
+			scale = 10486;
+			break;
+		case ICM42688_GYRO_FS_15_625:
+			scale = 20972;
+			break;
+		}
+	}
+
+	if (!is_accel && FIELD_GET(FIFO_HEADER_ACCEL, pkt[0]) == 1) {
+		offset += 7;
+	}
+
+	value = (int16_t)sys_le16_to_cpu((pkt[offset] << 8) | pkt[offset + 1]);
+
+	if (FIELD_GET(FIFO_HEADER_20, pkt[0]) == 1) {
+		uint32_t mask = is_accel ? GENMASK(7, 4) : GENMASK(3, 0);
+
+		offset = 0x11 + axis_offset;
+		value = (value << 4) | FIELD_GET(mask, pkt[offset]);
+		/* In 20 bit mode, FS can only be +/-16g and +/-2000dps */
+		scale = is_accel ? (INT64_C(16) * BIT(8) * 9.80665) : 131;
+		max = is_accel ? BIT(18) : BIT(19);
+		if (value == -524288) {
+			/* Invalid 20 bit value */
+			return -ENODATA;
+		}
+	} else {
+		if (value <= -32767) {
+			/* Invalid 16 bit value */
+			return -ENODATA;
+		}
+	}
+
+	*out = (q31_t)(value * scale / max);
+	return 0;
+}
+
+static uint32_t accel_period_ns[] = {
+	[ICM42688_ACCEL_ODR_1_5625] = UINT32_C(10000000000000) / 15625,
+	[ICM42688_ACCEL_ODR_3_125] = UINT32_C(10000000000000) / 31250,
+	[ICM42688_ACCEL_ODR_6_25] = UINT32_C(10000000000000) / 62500,
+	[ICM42688_ACCEL_ODR_12_5] = UINT32_C(10000000000000) / 12500,
+	[ICM42688_ACCEL_ODR_25] = UINT32_C(1000000000) / 25,
+	[ICM42688_ACCEL_ODR_50] = UINT32_C(1000000000) / 50,
+	[ICM42688_ACCEL_ODR_100] = UINT32_C(1000000000) / 100,
+	[ICM42688_ACCEL_ODR_200] = UINT32_C(1000000000) / 200,
+	[ICM42688_ACCEL_ODR_500] = UINT32_C(1000000000) / 500,
+	[ICM42688_ACCEL_ODR_1000] = UINT32_C(1000000),
+	[ICM42688_ACCEL_ODR_2000] = UINT32_C(1000000) / 2,
+	[ICM42688_ACCEL_ODR_4000] = UINT32_C(1000000) / 4,
+	[ICM42688_ACCEL_ODR_8000] = UINT32_C(1000000) / 8,
+	[ICM42688_ACCEL_ODR_16000] = UINT32_C(1000000) / 16,
+	[ICM42688_ACCEL_ODR_32000] = UINT32_C(1000000) / 32,
+};
+
+static uint32_t gyro_period_ns[] = {
+	[ICM42688_GYRO_ODR_12_5] = UINT32_C(10000000000000) / 12500,
+	[ICM42688_GYRO_ODR_25] = UINT32_C(1000000000) / 25,
+	[ICM42688_GYRO_ODR_50] = UINT32_C(1000000000) / 50,
+	[ICM42688_GYRO_ODR_100] = UINT32_C(1000000000) / 100,
+	[ICM42688_GYRO_ODR_200] = UINT32_C(1000000000) / 200,
+	[ICM42688_GYRO_ODR_500] = UINT32_C(1000000000) / 500,
+	[ICM42688_GYRO_ODR_1000] = UINT32_C(1000000),
+	[ICM42688_GYRO_ODR_2000] = UINT32_C(1000000) / 2,
+	[ICM42688_GYRO_ODR_4000] = UINT32_C(1000000) / 4,
+	[ICM42688_GYRO_ODR_8000] = UINT32_C(1000000) / 8,
+	[ICM42688_GYRO_ODR_16000] = UINT32_C(1000000) / 16,
+	[ICM42688_GYRO_ODR_32000] = UINT32_C(1000000) / 32,
+};
+
+static int icm42688_fifo_decode(const uint8_t *buffer, enum sensor_channel channel,
+				size_t channel_idx, uint32_t *fit, uint16_t max_count,
+				void *data_out)
+{
+	const struct icm42688_fifo_data *edata = (const struct icm42688_fifo_data *)buffer;
+	const uint8_t *buffer_end = buffer + sizeof(struct icm42688_fifo_data) + edata->fifo_count;
+	int accel_frame_count = 0;
+	int gyro_frame_count = 0;
+	int count = 0;
+	int rc;
+
+	if ((uintptr_t)buffer_end <= *fit || channel_idx != 0) {
+		return 0;
+	}
+
+	((struct sensor_data_header *)data_out)->base_timestamp_ns = edata->header.timestamp;
+
+	buffer += sizeof(struct icm42688_fifo_data);
+	while (count < max_count && buffer < buffer_end) {
+		const bool is_20b = FIELD_GET(FIFO_HEADER_20, buffer[0]) == 1;
+		const bool has_accel = FIELD_GET(FIFO_HEADER_ACCEL, buffer[0]) == 1;
+		const bool has_gyro = FIELD_GET(FIFO_HEADER_GYRO, buffer[0]) == 1;
+		const uint8_t *frame_end = buffer;
+
+		if (is_20b) {
+			frame_end += 20;
+		} else if (has_accel && has_gyro) {
+			frame_end += 16;
+		} else {
+			frame_end += 8;
+		}
+		if (has_accel) {
+			accel_frame_count++;
+		}
+		if (has_gyro) {
+			gyro_frame_count++;
+		}
+
+		if ((uintptr_t)buffer < *fit) {
+			/* This frame was already decoded, move on to the next frame */
+			buffer = frame_end;
+			continue;
+		}
+		if (channel == SENSOR_CHAN_DIE_TEMP) {
+			struct sensor_q31_data *data = (struct sensor_q31_data *)data_out;
+
+			data->shift = 9;
+			if (has_accel) {
+				data->readings[count].timestamp_delta =
+					accel_period_ns[edata->accel_odr] * (accel_frame_count - 1);
+			} else {
+				data->readings[count].timestamp_delta =
+					gyro_period_ns[edata->gyro_odr] * (gyro_frame_count - 1);
+			}
+			data->readings[count].temperature =
+				icm42688_read_temperature_from_packet(buffer);
+		} else if (IS_ACCEL(channel) && has_accel) {
+			/* Decode accel */
+			struct sensor_three_axis_data *data =
+				(struct sensor_three_axis_data *)data_out;
+			uint64_t period_ns = accel_period_ns[edata->accel_odr];
+
+			icm42688_get_shift(SENSOR_CHAN_ACCEL_XYZ, edata->header.accel_fs,
+					   edata->header.gyro_fs, &data->shift);
+
+			data->readings[count].timestamp_delta = (accel_frame_count - 1) * period_ns;
+			rc = icm42688_read_imu_from_packet(buffer, true, edata->header.accel_fs, 0,
+							   &data->readings[count].x);
+			rc |= icm42688_read_imu_from_packet(buffer, true, edata->header.accel_fs, 1,
+							    &data->readings[count].y);
+			rc |= icm42688_read_imu_from_packet(buffer, true, edata->header.accel_fs, 2,
+							    &data->readings[count].z);
+			if (rc != 0) {
+				accel_frame_count--;
+				buffer = frame_end;
+				continue;
+			}
+		} else if (IS_GYRO(channel) && has_gyro) {
+			/* Decode gyro */
+			struct sensor_three_axis_data *data =
+				(struct sensor_three_axis_data *)data_out;
+			uint64_t period_ns = accel_period_ns[edata->gyro_odr];
+
+			icm42688_get_shift(SENSOR_CHAN_GYRO_XYZ, edata->header.accel_fs,
+					   edata->header.gyro_fs, &data->shift);
+
+			data->readings[count].timestamp_delta = (gyro_frame_count - 1) * period_ns;
+			rc = icm42688_read_imu_from_packet(buffer, false, edata->header.gyro_fs, 0,
+							   &data->readings[count].x);
+			rc |= icm42688_read_imu_from_packet(buffer, false, edata->header.gyro_fs, 1,
+							    &data->readings[count].y);
+			rc |= icm42688_read_imu_from_packet(buffer, false, edata->header.gyro_fs, 2,
+							    &data->readings[count].z);
+			if (rc != 0) {
+				gyro_frame_count--;
+				buffer = frame_end;
+				continue;
+			}
+		}
+		buffer = frame_end;
+		*fit = (uintptr_t)frame_end;
+		count++;
+	}
+	return count;
+}
+
 static int icm42688_one_shot_decode(const uint8_t *buffer, enum sensor_channel channel,
-				   size_t channel_idx, uint32_t *fit,
-				   uint16_t max_count, void *data_out)
+				    size_t channel_idx, uint32_t *fit, uint16_t max_count,
+				    void *data_out)
 {
 	const struct icm42688_encoded_data *edata = (const struct icm42688_encoded_data *)buffer;
 	const struct icm42688_decoder_header *header = &edata->header;
@@ -318,34 +571,76 @@ static int icm42688_one_shot_decode(const uint8_t *buffer, enum sensor_channel c
 }
 
 static int icm42688_decoder_decode(const uint8_t *buffer, enum sensor_channel channel,
-				   size_t channel_idx, uint32_t *fit,
-				   uint16_t max_count, void *data_out)
+				   size_t channel_idx, uint32_t *fit, uint16_t max_count,
+				   void *data_out)
 {
+	const struct icm42688_decoder_header *header =
+		(const struct icm42688_decoder_header *)buffer;
+
+	if (header->is_fifo) {
+		return icm42688_fifo_decode(buffer, channel, channel_idx, fit, max_count, data_out);
+	}
 	return icm42688_one_shot_decode(buffer, channel, channel_idx, fit, max_count, data_out);
 }
 
 static int icm42688_decoder_get_frame_count(const uint8_t *buffer, enum sensor_channel channel,
 					    size_t channel_idx, uint16_t *frame_count)
 {
-	ARG_UNUSED(buffer);
+	const struct icm42688_fifo_data *data = (const struct icm42688_fifo_data *)buffer;
+	const struct icm42688_decoder_header *header = &data->header;
+
 	if (channel_idx != 0) {
 		return -ENOTSUP;
 	}
-	switch (channel) {
-	case SENSOR_CHAN_ACCEL_X:
-	case SENSOR_CHAN_ACCEL_Y:
-	case SENSOR_CHAN_ACCEL_Z:
-	case SENSOR_CHAN_ACCEL_XYZ:
-	case SENSOR_CHAN_GYRO_X:
-	case SENSOR_CHAN_GYRO_Y:
-	case SENSOR_CHAN_GYRO_Z:
-	case SENSOR_CHAN_GYRO_XYZ:
-	case SENSOR_CHAN_DIE_TEMP:
-		*frame_count = 1;
+
+	if (!header->is_fifo) {
+		switch (channel) {
+		case SENSOR_CHAN_ACCEL_X:
+		case SENSOR_CHAN_ACCEL_Y:
+		case SENSOR_CHAN_ACCEL_Z:
+		case SENSOR_CHAN_ACCEL_XYZ:
+		case SENSOR_CHAN_GYRO_X:
+		case SENSOR_CHAN_GYRO_Y:
+		case SENSOR_CHAN_GYRO_Z:
+		case SENSOR_CHAN_GYRO_XYZ:
+		case SENSOR_CHAN_DIE_TEMP:
+			*frame_count = 1;
+			return 0;
+		default:
+			return -ENOTSUP;
+		}
 		return 0;
-	default:
-		return -ENOTSUP;
 	}
+
+	/* Skip the header */
+	buffer += sizeof(struct icm42688_fifo_data);
+
+	uint16_t count = 0;
+	const uint8_t *end = buffer + data->fifo_count;
+
+	while (buffer < end) {
+		bool is_20b = FIELD_GET(FIFO_HEADER_20, buffer[0]);
+		int size = is_20b ? 3 : 2;
+
+		if (FIELD_GET(FIFO_HEADER_ACCEL, buffer[0])) {
+			size += 6;
+		}
+		if (FIELD_GET(FIFO_HEADER_GYRO, buffer[0])) {
+			size += 6;
+		}
+		if (FIELD_GET(FIFO_HEADER_TIMESTAMP_FSYNC, buffer[0])) {
+			size += 2;
+		}
+		if (is_20b) {
+			size += 3;
+		}
+
+		buffer += size;
+		++count;
+	}
+
+	*frame_count = count;
+	return 0;
 }
 
 static int icm42688_decoder_get_size_info(enum sensor_channel channel, size_t *base_size,
@@ -372,10 +667,31 @@ static int icm42688_decoder_get_size_info(enum sensor_channel channel, size_t *b
 	}
 }
 
+static bool icm24688_decoder_has_trigger(const uint8_t *buffer, enum sensor_trigger_type trigger)
+{
+	const struct icm42688_fifo_data *edata = (const struct icm42688_fifo_data *)buffer;
+
+	if (!edata->header.is_fifo) {
+		return false;
+	}
+
+	switch (trigger) {
+	case SENSOR_TRIG_DATA_READY:
+		return FIELD_GET(BIT_INT_STATUS_DATA_RDY, edata->int_status);
+	case SENSOR_TRIG_FIFO_WATERMARK:
+		return FIELD_GET(BIT_INT_STATUS_FIFO_THS, edata->int_status);
+	case SENSOR_TRIG_FIFO_FULL:
+		return FIELD_GET(BIT_INT_STATUS_FIFO_FULL, edata->int_status);
+	default:
+		return false;
+	}
+}
+
 SENSOR_DECODER_API_DT_DEFINE() = {
 	.get_frame_count = icm42688_decoder_get_frame_count,
 	.get_size_info = icm42688_decoder_get_size_info,
 	.decode = icm42688_decoder_decode,
+	.has_trigger = icm24688_decoder_has_trigger,
 };
 
 int icm42688_get_decoder(const struct device *dev, const struct sensor_decoder_api **decoder)
diff --git a/drivers/sensor/icm42688/icm42688_decoder.h b/drivers/sensor/icm42688/icm42688_decoder.h
index 2ef806373c9402..499cf3d0801554 100644
--- a/drivers/sensor/icm42688/icm42688_decoder.h
+++ b/drivers/sensor/icm42688/icm42688_decoder.h
@@ -18,6 +18,15 @@ struct icm42688_decoder_header {
 	uint8_t reserved: 2;
 } __attribute__((__packed__));
 
+struct icm42688_fifo_data {
+	struct icm42688_decoder_header header;
+	uint8_t int_status;
+	uint16_t gyro_odr: 4;
+	uint16_t accel_odr: 4;
+	uint16_t fifo_count: 11;
+	uint16_t reserved: 5;
+} __attribute__((__packed__));
+
 struct icm42688_encoded_data {
 	struct icm42688_decoder_header header;
 	struct {
diff --git a/drivers/sensor/icm42688/icm42688_reg.h b/drivers/sensor/icm42688/icm42688_reg.h
index dfc348c072c8c7..5f3b58f0b3c074 100644
--- a/drivers/sensor/icm42688/icm42688_reg.h
+++ b/drivers/sensor/icm42688/icm42688_reg.h
@@ -286,4 +286,12 @@
 #define MCLK_POLL_ATTEMPTS    100
 #define SOFT_RESET_TIME_MS    2 /* 1ms + elbow room */
 
+/* FIFO header */
+#define FIFO_HEADER_ACCEL BIT(6)
+#define FIFO_HEADER_GYRO BIT(5)
+#define FIFO_HEADER_20 BIT(4)
+#define FIFO_HEADER_TIMESTAMP_FSYNC GENMASK(3, 2)
+#define FIFO_HEADER_ODR_ACCEL BIT(1)
+#define FIFO_HEADER_ODR_GYRO BIT(0)
+
 #endif /* ZEPHYR_DRIVERS_SENSOR_ICM42688_REG_H_ */
diff --git a/drivers/sensor/icm42688/icm42688_rtio.c b/drivers/sensor/icm42688/icm42688_rtio.c
index c1b20689393d6c..ccb9532ee45a2c 100644
--- a/drivers/sensor/icm42688/icm42688_rtio.c
+++ b/drivers/sensor/icm42688/icm42688_rtio.c
@@ -8,6 +8,7 @@
 #include "icm42688.h"
 #include "icm42688_decoder.h"
 #include "icm42688_reg.h"
+#include "icm42688_rtio.h"
 #include "icm42688_spi.h"
 
 #include <zephyr/logging/log.h>
@@ -42,7 +43,7 @@ static int icm42688_rtio_sample_fetch(const struct device *dev, int16_t readings
 	return 0;
 }
 
-int icm42688_submit(const struct device *dev, struct rtio_iodev_sqe *iodev_sqe)
+static int icm42688_submit_one_shot(const struct device *dev, struct rtio_iodev_sqe *iodev_sqe)
 {
 	const struct sensor_read_config *cfg = iodev_sqe->sqe.iodev->data;
 	const enum sensor_channel *const channels = cfg->channels;
@@ -78,4 +79,17 @@ int icm42688_submit(const struct device *dev, struct rtio_iodev_sqe *iodev_sqe)
 	return 0;
 }
 
+int icm42688_submit(const struct device *dev, struct rtio_iodev_sqe *iodev_sqe)
+{
+	const struct sensor_read_config *cfg = iodev_sqe->sqe.iodev->data;
+
+	if (!cfg->is_streaming) {
+		return icm42688_submit_one_shot(dev, iodev_sqe);
+	} else if (IS_ENABLED(CONFIG_ICM42688_STREAM)) {
+		return icm42688_submit_stream(dev, iodev_sqe);
+	} else {
+		return -ENOTSUP;
+	}
+}
+
 BUILD_ASSERT(sizeof(struct icm42688_decoder_header) == 9);
diff --git a/drivers/sensor/icm42688/icm42688_rtio.h b/drivers/sensor/icm42688/icm42688_rtio.h
index 748c8a022b0923..888e8e95357aea 100644
--- a/drivers/sensor/icm42688/icm42688_rtio.h
+++ b/drivers/sensor/icm42688/icm42688_rtio.h
@@ -7,6 +7,13 @@
 #ifndef ZEPHYR_DRIVERS_SENSOR_ICM42688_RTIO_H_
 #define ZEPHYR_DRIVERS_SENSOR_ICM42688_RTIO_H_
 
+#include <zephyr/device.h>
+#include <zephyr/rtio/rtio.h>
+
 int icm42688_submit(const struct device *sensor, struct rtio_iodev_sqe *iodev_sqe);
 
+int icm42688_submit_stream(const struct device *sensor, struct rtio_iodev_sqe *iodev_sqe);
+
+void icm42688_fifo_event(const struct device *dev);
+
 #endif /* ZEPHYR_DRIVERS_SENSOR_ICM42688_RTIO_H_ */
diff --git a/drivers/sensor/icm42688/icm42688_rtio_stream.c b/drivers/sensor/icm42688/icm42688_rtio_stream.c
new file mode 100644
index 00000000000000..57d1a66cfde9cc
--- /dev/null
+++ b/drivers/sensor/icm42688/icm42688_rtio_stream.c
@@ -0,0 +1,319 @@
+/*
+ * Copyright (c) 2023 Google LLC
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <zephyr/logging/log.h>
+
+#include "icm42688.h"
+#include "icm42688_decoder.h"
+#include "icm42688_reg.h"
+#include "icm42688_rtio.h"
+
+LOG_MODULE_DECLARE(ICM42688_RTIO);
+
+int icm42688_submit_stream(const struct device *sensor, struct rtio_iodev_sqe *iodev_sqe)
+{
+	const struct sensor_read_config *cfg = iodev_sqe->sqe.iodev->data;
+	struct icm42688_dev_data *data = sensor->data;
+	struct icm42688_cfg new_config = data->cfg;
+
+	new_config.interrupt1_drdy = false;
+	new_config.interrupt1_fifo_ths = false;
+	new_config.interrupt1_fifo_full = false;
+	for (int i = 0; i < cfg->count; ++i) {
+		switch (cfg->triggers[i].trigger) {
+		case SENSOR_TRIG_DATA_READY:
+			new_config.interrupt1_drdy = true;
+			break;
+		case SENSOR_TRIG_FIFO_WATERMARK:
+			new_config.interrupt1_fifo_ths = true;
+			break;
+		case SENSOR_TRIG_FIFO_FULL:
+			new_config.interrupt1_fifo_full = true;
+			break;
+		default:
+			LOG_DBG("Trigger (%d) not supported", cfg->triggers[i].trigger);
+			break;
+		}
+	}
+
+	if (new_config.interrupt1_drdy != data->cfg.interrupt1_drdy ||
+	    new_config.interrupt1_fifo_ths != data->cfg.interrupt1_fifo_ths ||
+	    new_config.interrupt1_fifo_full != data->cfg.interrupt1_fifo_full) {
+		int rc = icm42688_safely_configure(sensor, &new_config);
+
+		if (rc != 0) {
+			LOG_ERR("Failed to configure sensor");
+			return rc;
+		}
+	}
+
+	data->streaming_sqe = iodev_sqe;
+	return 0;
+}
+
+static void icm42688_complete_cb(struct rtio *r, const struct rtio_sqe *sqe, void *arg)
+{
+	const struct device *dev = arg;
+	struct icm42688_dev_data *drv_data = dev->data;
+	const struct icm42688_dev_cfg *drv_cfg = dev->config;
+	struct rtio_iodev_sqe *iodev_sqe = sqe->userdata;
+
+	rtio_iodev_sqe_ok(iodev_sqe, drv_data->fifo_count);
+
+	gpio_pin_interrupt_configure_dt(&drv_cfg->gpio_int1, GPIO_INT_EDGE_TO_ACTIVE);
+}
+
+static void icm42688_fifo_count_cb(struct rtio *r, const struct rtio_sqe *sqe, void *arg)
+{
+	const struct device *dev = arg;
+	struct icm42688_dev_data *drv_data = dev->data;
+	const struct icm42688_dev_cfg *drv_cfg = dev->config;
+	struct rtio_iodev *spi_iodev = drv_data->spi_iodev;
+	uint8_t *fifo_count_buf = (uint8_t *)&drv_data->fifo_count;
+	uint16_t fifo_count = ((fifo_count_buf[0] << 8) | fifo_count_buf[1]);
+
+	drv_data->fifo_count = fifo_count;
+
+	/* Pull a operation from our device iodev queue, validated to only be reads */
+	struct rtio_iodev_sqe *iodev_sqe = drv_data->streaming_sqe;
+
+	drv_data->streaming_sqe = NULL;
+
+	/* Not inherently an underrun/overrun as we may have a buffer to fill next time */
+	if (iodev_sqe == NULL) {
+		LOG_DBG("No pending SQE");
+		gpio_pin_interrupt_configure_dt(&drv_cfg->gpio_int1, GPIO_INT_EDGE_TO_ACTIVE);
+		return;
+	}
+
+	const size_t packet_size = drv_data->cfg.fifo_hires ? 20 : 16;
+	const size_t min_read_size = sizeof(struct icm42688_fifo_data) + packet_size;
+	const size_t ideal_read_size = sizeof(struct icm42688_fifo_data) + fifo_count;
+	uint8_t *buf;
+	uint32_t buf_len;
+
+	if (rtio_sqe_rx_buf(iodev_sqe, min_read_size, ideal_read_size, &buf, &buf_len) != 0) {
+		LOG_ERR("Failed to get buffer");
+		rtio_iodev_sqe_err(iodev_sqe, -ENOMEM);
+		return;
+	}
+	LOG_DBG("Requesting buffer [%u, %u] got %u", (unsigned int)min_read_size,
+		(unsigned int)ideal_read_size, buf_len);
+
+	/* Read FIFO and call back to rtio with rtio_sqe completion */
+	/* TODO is packet format even needed? the fifo has a header per packet
+	 * already
+	 */
+	struct icm42688_fifo_data hdr = {
+		.header = {
+			.is_fifo = true,
+			.gyro_fs = drv_data->cfg.gyro_fs,
+			.accel_fs = drv_data->cfg.accel_fs,
+			.timestamp = drv_data->timestamp,
+		},
+		.int_status = drv_data->int_status,
+		.gyro_odr = drv_data->cfg.gyro_odr,
+		.accel_odr = drv_data->cfg.accel_odr,
+	};
+	uint32_t buf_avail = buf_len;
+
+	memcpy(buf, &hdr, sizeof(hdr));
+	buf_avail -= sizeof(hdr);
+
+	uint32_t read_len = MIN(fifo_count, buf_avail);
+	uint32_t pkts = read_len / packet_size;
+
+	read_len = pkts * packet_size;
+	((struct icm42688_fifo_data *)buf)->fifo_count = read_len;
+
+	__ASSERT_NO_MSG(read_len % pkt_size == 0);
+
+	uint8_t *read_buf = buf + sizeof(hdr);
+
+	/* Flush out completions  */
+	struct rtio_cqe *cqe;
+
+	do {
+		cqe = rtio_cqe_consume(r);
+		if (cqe != NULL) {
+			rtio_cqe_release(r, cqe);
+		}
+	} while (cqe != NULL);
+
+	/* Setup new rtio chain to read the fifo data and report then check the
+	 * result
+	 */
+	struct rtio_sqe *write_fifo_addr = rtio_sqe_acquire(r);
+	struct rtio_sqe *read_fifo_data = rtio_sqe_acquire(r);
+	struct rtio_sqe *complete_op = rtio_sqe_acquire(r);
+	const uint8_t reg_addr = REG_SPI_READ_BIT | FIELD_GET(REG_ADDRESS_MASK, REG_FIFO_DATA);
+
+	rtio_sqe_prep_tiny_write(write_fifo_addr, spi_iodev, RTIO_PRIO_NORM, &reg_addr, 1, NULL);
+	write_fifo_addr->flags = RTIO_SQE_TRANSACTION;
+	rtio_sqe_prep_read(read_fifo_data, spi_iodev, RTIO_PRIO_NORM, read_buf, read_len,
+			   iodev_sqe);
+
+	rtio_sqe_prep_callback(complete_op, icm42688_complete_cb, (void *)dev, iodev_sqe);
+
+	rtio_submit(r, 0);
+}
+
+static struct sensor_stream_trigger *
+icm42688_get_read_config_trigger(const struct sensor_read_config *cfg,
+				 enum sensor_trigger_type trig)
+{
+	for (int i = 0; i < cfg->count; ++i) {
+		if (cfg->triggers[i].trigger == trig) {
+			return &cfg->triggers[i];
+		}
+	}
+	LOG_DBG("Unsupported trigger (%d)", trig);
+	return NULL;
+}
+
+static void icm42688_int_status_cb(struct rtio *r, const struct rtio_sqe *sqr, void *arg)
+{
+	const struct device *dev = arg;
+	struct icm42688_dev_data *drv_data = dev->data;
+	const struct icm42688_dev_cfg *drv_cfg = dev->config;
+	struct rtio_iodev *spi_iodev = drv_data->spi_iodev;
+	struct rtio_iodev_sqe *streaming_sqe = drv_data->streaming_sqe;
+	struct sensor_read_config *read_config;
+
+	if (streaming_sqe == NULL) {
+		return;
+	}
+
+	read_config = (struct sensor_read_config *)streaming_sqe->sqe.iodev->data;
+	__ASSERT_NO_MSG(read_config != NULL);
+
+	if (!read_config->is_streaming) {
+		/* Oops, not really configured for streaming data */
+		return;
+	}
+
+	struct sensor_stream_trigger *fifo_ths_cfg =
+		icm42688_get_read_config_trigger(read_config, SENSOR_TRIG_FIFO_WATERMARK);
+	bool has_fifo_ths_trig = fifo_ths_cfg != NULL &&
+				 FIELD_GET(BIT_INT_STATUS_FIFO_THS, drv_data->int_status) != 0;
+
+	struct sensor_stream_trigger *fifo_full_cfg =
+		icm42688_get_read_config_trigger(read_config, SENSOR_TRIG_FIFO_FULL);
+	bool has_fifo_full_trig = fifo_full_cfg != NULL &&
+				  FIELD_GET(BIT_INT_STATUS_FIFO_FULL, drv_data->int_status) != 0;
+
+	if (!has_fifo_ths_trig && !has_fifo_full_trig) {
+		gpio_pin_interrupt_configure_dt(&drv_cfg->gpio_int1, GPIO_INT_EDGE_TO_ACTIVE);
+		return;
+	}
+
+	/* Flush completions */
+	struct rtio_cqe *cqe;
+
+	do {
+		cqe = rtio_cqe_consume(r);
+		if (cqe != NULL) {
+			rtio_cqe_release(r, cqe);
+		}
+	} while (cqe != NULL);
+
+	enum sensor_stream_data_opt data_opt;
+
+	if (has_fifo_ths_trig && !has_fifo_full_trig) {
+		/* Only care about fifo threshold */
+		data_opt = fifo_ths_cfg->opt;
+	} else if (!has_fifo_ths_trig && has_fifo_full_trig) {
+		/* Only care about fifo full */
+		data_opt = fifo_full_cfg->opt;
+	} else {
+		/* Both fifo threshold and full */
+		data_opt = MIN(fifo_ths_cfg->opt, fifo_full_cfg->opt);
+	}
+
+	if (data_opt == SENSOR_STREAM_DATA_NOP || data_opt == SENSOR_STREAM_DATA_DROP) {
+		uint8_t *buf;
+		uint32_t buf_len;
+
+		/* Clear streaming_sqe since we're done with the call */
+		drv_data->streaming_sqe = NULL;
+		if (rtio_sqe_rx_buf(streaming_sqe, sizeof(struct icm42688_fifo_data),
+				    sizeof(struct icm42688_fifo_data), &buf, &buf_len) != 0) {
+			rtio_iodev_sqe_err(streaming_sqe, -ENOMEM);
+			return;
+		}
+
+		struct icm42688_fifo_data *data = (struct icm42688_fifo_data *)buf;
+
+		memset(buf, 0, buf_len);
+		data->header.timestamp = drv_data->timestamp;
+		data->int_status = drv_data->int_status;
+		data->fifo_count = 0;
+		rtio_iodev_sqe_ok(streaming_sqe, 0);
+		gpio_pin_interrupt_configure_dt(&drv_cfg->gpio_int1, GPIO_INT_EDGE_TO_ACTIVE);
+		if (data_opt == SENSOR_STREAM_DATA_DROP) {
+			/* Flush the FIFO */
+			struct rtio_sqe *write_signal_path_reset = rtio_sqe_acquire(r);
+			uint8_t write_buffer[] = {
+				FIELD_GET(REG_ADDRESS_MASK, REG_SIGNAL_PATH_RESET),
+				BIT_FIFO_FLUSH,
+			};
+
+			rtio_sqe_prep_tiny_write(write_signal_path_reset, spi_iodev, RTIO_PRIO_NORM,
+						 write_buffer, ARRAY_SIZE(write_buffer), NULL);
+			/* TODO Add a new flag for fire-and-forget so we don't have to block here */
+			rtio_submit(r, 1);
+			ARG_UNUSED(rtio_cqe_consume(r));
+		}
+		return;
+	}
+
+	/* We need the data, read the fifo length */
+	struct rtio_sqe *write_fifo_count_reg = rtio_sqe_acquire(r);
+	struct rtio_sqe *read_fifo_count = rtio_sqe_acquire(r);
+	struct rtio_sqe *check_fifo_count = rtio_sqe_acquire(r);
+	uint8_t reg = REG_SPI_READ_BIT | FIELD_GET(REG_ADDRESS_MASK, REG_FIFO_COUNTH);
+	uint8_t *read_buf = (uint8_t *)&drv_data->fifo_count;
+
+	rtio_sqe_prep_tiny_write(write_fifo_count_reg, spi_iodev, RTIO_PRIO_NORM, &reg, 1, NULL);
+	write_fifo_count_reg->flags = RTIO_SQE_TRANSACTION;
+	rtio_sqe_prep_read(read_fifo_count, spi_iodev, RTIO_PRIO_NORM, read_buf, 2, NULL);
+	rtio_sqe_prep_callback(check_fifo_count, icm42688_fifo_count_cb, arg, NULL);
+
+	rtio_submit(r, 0);
+}
+
+void icm42688_fifo_event(const struct device *dev)
+{
+	struct icm42688_dev_data *drv_data = dev->data;
+	struct rtio_iodev *spi_iodev = drv_data->spi_iodev;
+	struct rtio *r = drv_data->r;
+
+	if (drv_data->streaming_sqe == NULL) {
+		return;
+	}
+
+	drv_data->timestamp = k_ticks_to_ns_floor64(k_uptime_ticks());
+
+	/*
+	 * Setup rtio chain of ops with inline calls to make decisions
+	 * 1. read int status
+	 * 2. call to check int status and get pending RX operation
+	 * 4. read fifo len
+	 * 5. call to determine read len
+	 * 6. read fifo
+	 * 7. call to report completion
+	 */
+	struct rtio_sqe *write_int_reg = rtio_sqe_acquire(r);
+	struct rtio_sqe *read_int_reg = rtio_sqe_acquire(r);
+	struct rtio_sqe *check_int_status = rtio_sqe_acquire(r);
+	uint8_t reg = REG_SPI_READ_BIT | FIELD_GET(REG_ADDRESS_MASK, REG_INT_STATUS);
+
+	rtio_sqe_prep_tiny_write(write_int_reg, spi_iodev, RTIO_PRIO_NORM, &reg, 1, NULL);
+	write_int_reg->flags = RTIO_SQE_TRANSACTION;
+	rtio_sqe_prep_read(read_int_reg, spi_iodev, RTIO_PRIO_NORM, &drv_data->int_status, 1, NULL);
+	rtio_sqe_prep_callback(check_int_status, icm42688_int_status_cb, (void *)dev, NULL);
+	rtio_submit(r, 0);
+}
diff --git a/drivers/sensor/icm42688/icm42688_trigger.c b/drivers/sensor/icm42688/icm42688_trigger.c
index adaecf84baa656..32b60f3cd75945 100644
--- a/drivers/sensor/icm42688/icm42688_trigger.c
+++ b/drivers/sensor/icm42688/icm42688_trigger.c
@@ -12,6 +12,7 @@
 
 #include "icm42688.h"
 #include "icm42688_reg.h"
+#include "icm42688_rtio.h"
 #include "icm42688_spi.h"
 #include "icm42688_trigger.h"
 
@@ -30,6 +31,9 @@ static void icm42688_gpio_callback(const struct device *dev, struct gpio_callbac
 #elif defined(CONFIG_ICM42688_TRIGGER_GLOBAL_THREAD)
 	k_work_submit(&data->work);
 #endif
+	if (IS_ENABLED(CONFIG_ICM42688_STREAM)) {
+		icm42688_fifo_event(data->dev);
+	}
 }
 
 #if defined(CONFIG_ICM42688_TRIGGER_OWN_THREAD) || defined(CONFIG_ICM42688_TRIGGER_GLOBAL_THREAD)
@@ -90,6 +94,8 @@ int icm42688_trigger_set(const struct device *dev, const struct sensor_trigger *
 
 	switch (trig->type) {
 	case SENSOR_TRIG_DATA_READY:
+	case SENSOR_TRIG_FIFO_WATERMARK:
+	case SENSOR_TRIG_FIFO_FULL:
 		data->data_ready_handler = handler;
 		data->data_ready_trigger = trig;
 
@@ -146,7 +152,7 @@ int icm42688_trigger_init(const struct device *dev)
 	return gpio_pin_interrupt_configure_dt(&cfg->gpio_int1, GPIO_INT_EDGE_TO_ACTIVE);
 }
 
-int icm42688_trigger_enable_interrupt(const struct device *dev)
+int icm42688_trigger_enable_interrupt(const struct device *dev, struct icm42688_cfg *new_cfg)
 {
 	int res;
 	const struct icm42688_dev_cfg *cfg = dev->config;
@@ -164,9 +170,19 @@ int icm42688_trigger_enable_interrupt(const struct device *dev)
 		return res;
 	}
 
-	/* enable data ready interrupt on INT1 pin */
-	return icm42688_spi_single_write(&cfg->spi, REG_INT_SOURCE0,
-					 FIELD_PREP(BIT_UI_DRDY_INT1_EN, 1));
+	/* enable interrupts on INT1 pin */
+	uint8_t value = 0;
+
+	if (new_cfg->interrupt1_drdy) {
+		value |= FIELD_PREP(BIT_UI_DRDY_INT1_EN, 1);
+	}
+	if (new_cfg->interrupt1_fifo_ths) {
+		value |= FIELD_PREP(BIT_FIFO_THS_INT1_EN, 1);
+	}
+	if (new_cfg->interrupt1_fifo_full) {
+		value |= FIELD_PREP(BIT_FIFO_FULL_INT1_EN, 1);
+	}
+	return icm42688_spi_single_write(&cfg->spi, REG_INT_SOURCE0, value);
 }
 
 void icm42688_lock(const struct device *dev)
diff --git a/drivers/sensor/icm42688/icm42688_trigger.h b/drivers/sensor/icm42688/icm42688_trigger.h
index 5ed382eb0d403a..e0397591618120 100644
--- a/drivers/sensor/icm42688/icm42688_trigger.h
+++ b/drivers/sensor/icm42688/icm42688_trigger.h
@@ -26,9 +26,10 @@ int icm42688_trigger_init(const struct device *dev);
  * @brief enable the trigger gpio interrupt
  *
  * @param dev icm42688 device pointer
+ * @param new_cfg New configuration to use for the device
  * @return int 0 on success, negative error code otherwise
  */
-int icm42688_trigger_enable_interrupt(const struct device *dev);
+int icm42688_trigger_enable_interrupt(const struct device *dev, struct icm42688_cfg *new_cfg);
 
 /**
  * @brief lock access to the icm42688 device driver

From 80963ac2efa905d88c9b19b01950aaf6d1f015da Mon Sep 17 00:00:00 2001
From: Yuval Peress <peress@google.com>
Date: Tue, 10 Oct 2023 01:59:04 -0600
Subject: [PATCH 3/3] sensors: convert fifo_wm to batch_dur

Having a % FIFO watermark isn't very useful as it doesn't convey how long
the SoC can sleep (or do other work) while batching sensor data. Convert
the attribute to a batch duration using ticks. Currently the ticks are
in system ticks, but eventually when an external clock is attached to
the sensor it will be in the external clock's ticks.

Signed-off-by: Yuval Peress <peress@google.com>
---
 drivers/sensor/icm42688/icm42688.c        | 19 ++++---
 drivers/sensor/icm42688/icm42688.h        |  2 +-
 drivers/sensor/icm42688/icm42688_common.c | 61 ++++++++++++++++++++++-
 drivers/sensor/sensor_shell.c             |  2 +-
 include/zephyr/drivers/sensor.h           |  6 ++-
 5 files changed, 78 insertions(+), 12 deletions(-)

diff --git a/drivers/sensor/icm42688/icm42688.c b/drivers/sensor/icm42688/icm42688.c
index ee92c30987ecfc..a6b4cd06c52197 100644
--- a/drivers/sensor/icm42688/icm42688.c
+++ b/drivers/sensor/icm42688/icm42688.c
@@ -157,13 +157,11 @@ static int icm42688_attr_set(const struct device *dev, enum sensor_channel chan,
 		}
 		break;
 	case SENSOR_CHAN_ALL:
-		if (attr == SENSOR_ATTR_FIFO_WATERMARK) {
-			int64_t mval = sensor_value_to_micro(val);
-
-			if (mval < 0 || mval > 1000000) {
+		if (attr == SENSOR_ATTR_BATCH_DURATION) {
+			if (val->val1 < 0) {
 				return -EINVAL;
 			}
-			new_config.fifo_wm = CLAMP(mval * 2048 / 1000000, 0, 2048);
+			new_config.batch_ticks = val->val1;
 		} else {
 			LOG_ERR("Unsupported attribute");
 			res = -EINVAL;
@@ -217,6 +215,15 @@ static int icm42688_attr_get(const struct device *dev, enum sensor_channel chan,
 			res = -EINVAL;
 		}
 		break;
+	case SENSOR_CHAN_ALL:
+		if (attr == SENSOR_ATTR_BATCH_DURATION) {
+			val->val1 = cfg->batch_ticks;
+			val->val2 = 0;
+		} else {
+			LOG_ERR("Unsupported attribute");
+			res = -EINVAL;
+		}
+		break;
 	default:
 		LOG_ERR("Unsupported channel");
 		res = -EINVAL;
@@ -272,7 +279,7 @@ int icm42688_init(const struct device *dev)
 	data->cfg.gyro_odr = ICM42688_GYRO_ODR_1000;
 	data->cfg.temp_dis = false;
 	data->cfg.fifo_en = IS_ENABLED(CONFIG_ICM42688_STREAM);
-	data->cfg.fifo_wm = 0;
+	data->cfg.batch_ticks = 0;
 	data->cfg.fifo_hires = 0;
 	data->cfg.interrupt1_drdy = 0;
 	data->cfg.interrupt1_fifo_ths = 0;
diff --git a/drivers/sensor/icm42688/icm42688.h b/drivers/sensor/icm42688/icm42688.h
index c4167e6a3b9cf4..5ec470cff270c3 100644
--- a/drivers/sensor/icm42688/icm42688.h
+++ b/drivers/sensor/icm42688/icm42688.h
@@ -380,7 +380,7 @@ struct icm42688_cfg {
 	/* TODO timestamp options */
 
 	bool fifo_en;
-	uint16_t fifo_wm;
+	int32_t batch_ticks;
 	bool fifo_hires;
 	/* TODO additional FIFO options */
 
diff --git a/drivers/sensor/icm42688/icm42688_common.c b/drivers/sensor/icm42688/icm42688_common.c
index 9cb3037ed06ea3..4df2f787092766 100644
--- a/drivers/sensor/icm42688/icm42688_common.c
+++ b/drivers/sensor/icm42688/icm42688_common.c
@@ -62,6 +62,62 @@ int icm42688_reset(const struct device *dev)
 	return 0;
 }
 
+static uint16_t icm42688_compute_fifo_wm(const struct icm42688_cfg *cfg)
+{
+	const bool accel_enabled = cfg->accel_mode != ICM42688_ACCEL_OFF;
+	const bool gyro_enabled = cfg->gyro_mode != ICM42688_GYRO_OFF;
+	const int pkt_size = cfg->fifo_hires ? 20 : (accel_enabled && gyro_enabled ? 16 : 8);
+	int accel_modr = 0;
+	int gyro_modr = 0;
+	int64_t modr;
+
+	if (cfg->batch_ticks == 0 || (!accel_enabled && !gyro_enabled)) {
+		return 0;
+	}
+
+	if (accel_enabled) {
+		struct sensor_value val = {0};
+
+		icm42688_accel_reg_to_hz(cfg->accel_odr, &val);
+		accel_modr = sensor_value_to_micro(&val) / 1000;
+	}
+	if (gyro_enabled) {
+		struct sensor_value val = {0};
+
+		icm42688_gyro_reg_to_odr(cfg->gyro_odr, &val);
+		gyro_modr = sensor_value_to_micro(&val) / 1000;
+	}
+
+	if (accel_modr == 0) {
+		modr = gyro_modr;
+	} else if (gyro_modr == 0) {
+		modr = accel_modr;
+	} else {
+		/* Need to find the least common multiplier (LCM) */
+		int n1 = accel_modr;
+		int n2 = gyro_modr;
+
+		while (n1 != n2) {
+			if (n1 > n2) {
+				n1 -= n2;
+			} else {
+				n2 -= n1;
+			}
+		}
+		LOG_DBG("GCD=%d", n1);
+		modr = ((int64_t)accel_modr * (int64_t)gyro_modr) / n1;
+	}
+	/* At this point we have 'modr' as mHz which is 1 / msec. */
+
+	/* Convert 'modr' to bytes * batch_ticks / msec */
+	modr *= (int64_t)cfg->batch_ticks * pkt_size;
+
+	/* 'modr' = byte_ticks_per_msec / kticks_per_sec */
+	modr = DIV_ROUND_UP(modr, CONFIG_SYS_CLOCK_TICKS_PER_SEC * INT64_C(1000));
+
+	return (uint16_t)MIN(modr, 0x7ff);
+}
+
 int icm42688_configure(const struct device *dev, struct icm42688_cfg *cfg)
 {
 	struct icm42688_dev_data *dev_data = dev->data;
@@ -210,7 +266,8 @@ int icm42688_configure(const struct device *dev, struct icm42688_cfg *cfg)
 		}
 
 		/* Set watermark and interrupt handling first */
-		uint8_t fifo_wml = (cfg->fifo_wm) & 0xFF;
+		uint16_t fifo_wm = icm42688_compute_fifo_wm(cfg);
+		uint8_t fifo_wml = fifo_wm & 0xFF;
 
 		LOG_DBG("FIFO_CONFIG2( (0x%x)) (WM Low) 0x%x", REG_FIFO_CONFIG2, fifo_wml);
 		res = icm42688_spi_single_write(&dev_cfg->spi, REG_FIFO_CONFIG2, fifo_wml);
@@ -219,7 +276,7 @@ int icm42688_configure(const struct device *dev, struct icm42688_cfg *cfg)
 			return -EINVAL;
 		}
 
-		uint8_t fifo_wmh = (cfg->fifo_wm >> 8) & 0x0F;
+		uint8_t fifo_wmh = (fifo_wm >> 8) & 0x0F;
 
 		LOG_DBG("FIFO_CONFIG3 (0x%x) (WM High) 0x%x", REG_FIFO_CONFIG3, fifo_wmh);
 		res = icm42688_spi_single_write(&dev_cfg->spi, REG_FIFO_CONFIG3, fifo_wmh);
diff --git a/drivers/sensor/sensor_shell.c b/drivers/sensor/sensor_shell.c
index 5d41663c491db8..bfeac4e452fd6c 100644
--- a/drivers/sensor/sensor_shell.c
+++ b/drivers/sensor/sensor_shell.c
@@ -122,7 +122,7 @@ static const char *sensor_attribute_name[SENSOR_ATTR_COMMON_COUNT] = {
 	[SENSOR_ATTR_FEATURE_MASK] = "feature_mask",
 	[SENSOR_ATTR_ALERT] = "alert",
 	[SENSOR_ATTR_FF_DUR] = "ff_dur",
-	[SENSOR_ATTR_FIFO_WATERMARK] = "fifo_wm",
+	[SENSOR_ATTR_BATCH_DURATION] = "batch_dur",
 };
 
 /* Forward declaration */
diff --git a/include/zephyr/drivers/sensor.h b/include/zephyr/drivers/sensor.h
index 715d5ff5385975..ef688286164512 100644
--- a/include/zephyr/drivers/sensor.h
+++ b/include/zephyr/drivers/sensor.h
@@ -334,8 +334,10 @@ enum sensor_attribute {
 	 *  to the new sampling frequency.
 	 */
 	SENSOR_ATTR_FF_DUR,
-	/** Watermark % for the hardware fifo interrupt */
-	SENSOR_ATTR_FIFO_WATERMARK,
+
+	/** Hardware batch duration in ticks */
+	SENSOR_ATTR_BATCH_DURATION,
+
 	/**
 	 * Number of all common sensor attributes.
 	 */