From cd5c2b2d8423a364f799a2199eb232233395b408 Mon Sep 17 00:00:00 2001
From: Yuval Peress <peress@google.com>
Date: Mon, 5 Jun 2023 12:08:58 -0600
Subject: [PATCH] WIP streaming

Signed-off-by: Yuval Peress <peress@google.com>
---
 drivers/sensor/Kconfig                        |   9 +
 drivers/sensor/default_rtio_sensor.c          |   4 +-
 drivers/sensor/icm42688/CMakeLists.txt        |   1 +
 drivers/sensor/icm42688/Kconfig               |   9 +
 drivers/sensor/icm42688/icm42688.c            |  13 +
 drivers/sensor/icm42688/icm42688.h            |  12 +
 drivers/sensor/icm42688/icm42688_common.c     |   9 +-
 drivers/sensor/icm42688/icm42688_decoder.c    | 157 ++++++++-
 drivers/sensor/icm42688/icm42688_decoder.h    |   9 +
 drivers/sensor/icm42688/icm42688_reg.h        |   8 +
 drivers/sensor/icm42688/icm42688_rtio.c       | 229 +++++++++++-
 drivers/sensor/icm42688/icm42688_rtio.h       |   2 +
 drivers/sensor/icm42688/icm42688_trigger.c    |  24 +-
 drivers/sensor/icm42688/icm42688_trigger.h    |   3 +-
 drivers/sensor/sensor_shell.c                 | 331 +++++++++++++++---
 include/zephyr/drivers/sensor.h               |  81 ++++-
 .../sensor_shell/boards/tdk_robokit1.conf     |   3 +
 17 files changed, 839 insertions(+), 65 deletions(-)
 create mode 100644 samples/sensor/sensor_shell/boards/tdk_robokit1.conf

diff --git a/drivers/sensor/Kconfig b/drivers/sensor/Kconfig
index 838bdf1a988ce68..01d4be0c556eba9 100644
--- a/drivers/sensor/Kconfig
+++ b/drivers/sensor/Kconfig
@@ -36,6 +36,15 @@ config SENSOR_SHELL
 	help
 	  This shell provides access to basic sensor data.
 
+config SENSOR_SHELL_THREAD_STACK_SIZE
+	int "Stack size for the sensor shell data processing thread"
+	depends on SENSOR_SHELL
+	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 55db893f4d89c04..a86bd81ec1d48fe 100644
--- a/drivers/sensor/default_rtio_sensor.c
+++ b/drivers/sensor/default_rtio_sensor.c
@@ -22,8 +22,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);
 	}
 }
 
diff --git a/drivers/sensor/icm42688/CMakeLists.txt b/drivers/sensor/icm42688/CMakeLists.txt
index cf9308bb95c4478..4bc7db3b6173732 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.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 23812c66d25dd59..8318ec7bd1b27e7 100644
--- a/drivers/sensor/icm42688/Kconfig
+++ b/drivers/sensor/icm42688/Kconfig
@@ -50,6 +50,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 de7780698278777..16cf1deb389b4c4 100644
--- a/drivers/sensor/icm42688/icm42688.c
+++ b/drivers/sensor/icm42688/icm42688.c
@@ -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;
diff --git a/drivers/sensor/icm42688/icm42688.h b/drivers/sensor/icm42688/icm42688.h
index 18ae4606e72593a..c4167e6a3b9cf4b 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 6e82f22678bd3e9..b787ce5756fda2e 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);
@@ -173,8 +174,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 582e7ac52b1e9dc..ffb397e6454ea39 100644
--- a/drivers/sensor/icm42688/icm42688_decoder.c
+++ b/drivers/sensor/icm42688/icm42688_decoder.c
@@ -214,6 +214,98 @@ int icm42688_encode(const struct device *dev, const enum sensor_channel *const c
 	return 0;
 }
 
+static int icm42688_fifo_decode(const uint8_t *buffer, sensor_frame_iterator_t *fit,
+				sensor_channel_iterator_t *cit, enum sensor_channel *channels,
+				q31_t *values, uint8_t max_count)
+{
+	const struct icm42688_fifo_data *edata = (const struct icm42688_fifo_data *)buffer;
+
+	if (edata->fifo_count <= *fit) {
+		return 0;
+	}
+
+	/* Jump to the frame start */
+	buffer += sizeof(struct icm42688_fifo_data) + *fit;
+
+	const bool is_20b = FIELD_GET(FIFO_HEADER_20, buffer[0]);
+	const bool has_accel = FIELD_GET(FIFO_HEADER_ACCEL, buffer[0]);
+	const bool has_gyro = FIELD_GET(FIFO_HEADER_GYRO, buffer[0]);
+	const int channel_count = 1 + (has_accel ? 3 : 0) + (has_gyro ? 3 : 0);
+	const uint8_t *frame_end = buffer + /* header */ 1 + /* accel */ (has_accel ? 6 : 0) +
+				   /* gyro */ (has_gyro ? 6 : 0) +
+				   /* timestamp */ (has_accel && has_gyro ? 2 : 0) +
+				   /* temperature */ (is_20b ? 2 : 1);
+	int header_channels[7];
+	int count = 0;
+
+	if (*cit >= channel_count) {
+		return -1;
+	}
+
+	if (has_accel) {
+		header_channels[count++] = SENSOR_CHAN_ACCEL_X;
+		header_channels[count++] = SENSOR_CHAN_ACCEL_Y;
+		header_channels[count++] = SENSOR_CHAN_ACCEL_Z;
+	}
+	if (has_gyro) {
+		header_channels[count++] = SENSOR_CHAN_GYRO_X;
+		header_channels[count++] = SENSOR_CHAN_GYRO_Y;
+		header_channels[count++] = SENSOR_CHAN_GYRO_Z;
+	}
+	header_channels[count] = SENSOR_CHAN_DIE_TEMP;
+	count = 0;
+
+	while (count < max_count && *cit < channel_count) {
+		/*
+		 * [0]@0 - accel_x
+		 * [1]@2 - accel_y
+		 * [2]@4 - accel_z
+		 * [3]@6 - gyro_x
+		 * [4]@8 - gyro_y
+		 * [5]@a - gyro_z
+		 * [6]@c - temp
+		 */
+		int offset = 1 + *cit * 2;
+
+		channels[count] = header_channels[*cit];
+		int32_t value;
+		int32_t value_max = INT16_MAX;
+
+		if (channels[count] == SENSOR_CHAN_DIE_TEMP) {
+			value = buffer[offset];
+			if (is_20b) {
+				value |= (buffer[offset + 1] << 8);
+			} else {
+				value_max = INT8_MAX;
+			}
+		} else {
+			value = (buffer[offset] << 8) | buffer[offset + 1];
+			if (is_20b) {
+				int mask = channels[count] < SENSOR_CHAN_ACCEL_XYZ ? GENMASK(7, 4)
+										   : GENMASK(3, 0);
+				/* Get the offset after the end of the frame data */
+				int ext_offset =
+					(channels[count] - (channels[count] < SENSOR_CHAN_ACCEL_XYZ
+								    ? SENSOR_CHAN_ACCEL_X
+								    : SENSOR_CHAN_GYRO_X));
+
+				value = (value << 4) | FIELD_GET(mask, frame_end[ext_offset]);
+				value_max = (1 << 19) - 1;
+			}
+		}
+		values[count] = (q31_t)((int64_t)value * INT32_MAX / value_max);
+		count++;
+		*cit += 1;
+	}
+
+	if (*cit == channel_count) {
+		/* Reached the end of the frame */
+		*fit += (uintptr_t)frame_end - (uintptr_t)buffer + (is_20b ? 3 : 0);
+		*cit = 0;
+	}
+	return count;
+}
+
 static int icm42688_one_shot_decode(const uint8_t *buffer, sensor_frame_iterator_t *fit,
 				    sensor_channel_iterator_t *cit, enum sensor_channel *channels,
 				    q31_t *values, uint8_t max_count)
@@ -267,13 +359,53 @@ static int icm42688_decoder_decode(const uint8_t *buffer, sensor_frame_iterator_
 				   sensor_channel_iterator_t *cit, enum sensor_channel *channels,
 				   q31_t *values, uint8_t max_count)
 {
+	const struct icm42688_decoder_header *header =
+		(const struct icm42688_decoder_header *)buffer;
+
+	if (header->is_fifo) {
+		return icm42688_fifo_decode(buffer, fit, cit, channels, values, max_count);
+	}
 	return icm42688_one_shot_decode(buffer, fit, cit, channels, values, max_count);
 }
 
 static int icm42688_decoder_get_frame_count(const uint8_t *buffer, uint16_t *frame_count)
 {
-	ARG_UNUSED(buffer);
-	*frame_count = 1;
+	const struct icm42688_fifo_data *data = (const struct icm42688_fifo_data *)buffer;
+	const struct icm42688_decoder_header *header = &data->header;
+
+	if (!header->is_fifo) {
+		*frame_count = 1;
+		return 0;
+	}
+
+	/* 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;
 }
 
@@ -286,6 +418,26 @@ static int icm42688_decoder_get_timestamp(const uint8_t *buffer, uint64_t *times
 	return 0;
 }
 
+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_THRESHOLD:
+		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;
+	}
+}
+
 static int icm42688_decoder_get_shift(const uint8_t *buffer, enum sensor_channel channel_type,
 				      int8_t *shift)
 {
@@ -299,6 +451,7 @@ SENSOR_DECODER_API_DT_DEFINE() = {
 	.get_frame_count = icm42688_decoder_get_frame_count,
 	.get_timestamp = icm42688_decoder_get_timestamp,
 	.get_shift = icm42688_decoder_get_shift,
+	.has_trigger = icm24688_decoder_has_trigger,
 	.decode = icm42688_decoder_decode,
 };
 
diff --git a/drivers/sensor/icm42688/icm42688_decoder.h b/drivers/sensor/icm42688/icm42688_decoder.h
index 402ba82d3d59fa4..a9462975ce64783 100644
--- a/drivers/sensor/icm42688/icm42688_decoder.h
+++ b/drivers/sensor/icm42688/icm42688_decoder.h
@@ -18,6 +18,15 @@ struct icm42688_decoder_header {
 	uint64_t timestamp;
 } __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 dfc348c072c8c75..5f3b58f0b3c074d 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 c1b20689393d6c0..00193c581d8e9a5 100644
--- a/drivers/sensor/icm42688/icm42688_rtio.c
+++ b/drivers/sensor/icm42688/icm42688_rtio.c
@@ -42,7 +42,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 +78,231 @@ 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)
+{
+	struct icm42688_dev_data *data = dev->data;
+	const struct sensor_read_config *cfg = iodev_sqe->sqe.iodev->data;
+	int rc;
+
+	if (!cfg->is_streaming) {
+		return icm42688_submit_one_shot(dev, iodev_sqe);
+	}
+
+	/* Only support */
+	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_THRESHOLD:
+			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) {
+		rc = icm42688_safely_configure(dev, &new_config);
+		if (rc != 0) {
+			LOG_ERR("Failed to configure sensor");
+			return rc;
+		}
+	}
+
+	data->streaming_sqe = iodev_sqe;
+	return 0;
+}
+
 BUILD_ASSERT(sizeof(struct icm42688_decoder_header) == 9);
+
+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;
+
+	/* TODO report number of frames or bytes read here */
+	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 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;
+
+	if (FIELD_GET(BIT_INT_STATUS_FIFO_THS, drv_data->int_status) == 0 &&
+	    FIELD_GET(BIT_INT_STATUS_FIFO_FULL, drv_data->int_status) == 0) {
+		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);
+
+	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_rtio.h b/drivers/sensor/icm42688/icm42688_rtio.h
index 748c8a022b09238..9b236aa41f73e25 100644
--- a/drivers/sensor/icm42688/icm42688_rtio.h
+++ b/drivers/sensor/icm42688/icm42688_rtio.h
@@ -9,4 +9,6 @@
 
 int icm42688_submit(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_trigger.c b/drivers/sensor/icm42688/icm42688_trigger.c
index adaecf84baa6566..b00df2596bcabd2 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_THRESHOLD:
+	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 5ed382eb0d403a1..e03975916181203 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
diff --git a/drivers/sensor/sensor_shell.c b/drivers/sensor/sensor_shell.c
index b8c828297318fd5..5e531450aaf8f10 100644
--- a/drivers/sensor/sensor_shell.c
+++ b/drivers/sensor/sensor_shell.c
@@ -21,6 +21,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> .. "                                   \
@@ -36,7 +41,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",
@@ -94,6 +99,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] = {
@@ -112,6 +118,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 */
@@ -144,12 +151,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_THRESHOLD, 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;
@@ -158,12 +185,25 @@ static enum dynamic_command_context current_cmd_ctx = NONE;
 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),
 };
 RTIO_IODEV_DEFINE(iodev_sensor_shell_read, &__sensor_iodev_api, &iodev_sensor_shell_read_config);
 
+/* 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);
+
 /* Create the RTIO context to service the reading */
 RTIO_DEFINE_WITH_MEMPOOL(sensor_read_rtio, 8, 8, 32, 64, 4);
 
@@ -234,14 +274,43 @@ struct sensor_shell_processing_context {
 	const struct shell *sh;
 };
 
+static void sensor_shell_print_q_value(const struct shell *sh, int channel, int64_t value, int8_t shift)
+{
+	int64_t scaled_value = (shift < 0) ? (value >> -shift) : (value << shift);
+	bool is_negative = scaled_value < 0;
+	int numerator;
+	int denominator;
+
+	scaled_value = llabs(scaled_value);
+	numerator = (int)FIELD_GET(GENMASK64(31 + shift, 31), scaled_value);
+	denominator = (int)((FIELD_GET(GENMASK64(30, 0), scaled_value) * 1000000) / INT32_MAX);
+
+	if (channel >= ARRAY_SIZE(sensor_channel_name)) {
+		shell_print(sh, "channel idx=%d shift=%d value=%s%d.%06d", channel, shift,
+			    is_negative ? "-" : "", numerator, denominator);
+	} else {
+		shell_print(sh, "channel idx=%d %s shift=%d value=%s%d.%06d", channel,
+			    sensor_channel_name[channel], shift, is_negative ? "-" : "", numerator,
+			    denominator);
+	}
+}
+
 static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t buf_len,
 					     void *userdata)
 {
+	static struct {
+		int64_t accumulator;
+		int8_t shift;
+		int count;
+		bool is_active;
+	} sample_stats[SENSOR_CHAN_COMMON_COUNT];
+
 	struct sensor_shell_processing_context *ctx = userdata;
 	const struct sensor_decoder_api *decoder;
 	sensor_frame_iterator_t fit = {0};
 	sensor_channel_iterator_t cit = {0};
 	uint64_t timestamp;
+	uint16_t frame_count;
 	enum sensor_channel channel;
 	q31_t q;
 	int rc;
@@ -264,7 +333,24 @@ static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t
 		shell_error(ctx->sh, "Failed to get fetch timestamp for '%s'", ctx->dev->name);
 		return;
 	}
-	shell_print(ctx->sh, "Got samples at %" PRIu64 " ns", timestamp);
+	rc = decoder->get_frame_count(buf, &frame_count);
+	if (rc != 0) {
+		shell_error(ctx->sh, "Failed to get frame count for '%s'", ctx->dev->name);
+		return;
+	}
+	shell_print(ctx->sh, "Got %" PRIu16 " frame(s) at %" PRIu64 " ns", frame_count, timestamp);
+
+	for (int i = 0; i < ARRAY_SIZE(sensor_trigger_table); ++i) {
+		bool has_trigger = decoder->has_trigger(buf, sensor_trigger_table[i].trigger.type);
+
+		if (has_trigger) {
+			shell_print(ctx->sh, "Trigger '%s' detected", sensor_trigger_table[i].name);
+		}
+	}
+
+	if (frame_count > 1) {
+		memset(sample_stats, 0, sizeof(sample_stats));
+	}
 
 	while (decoder->decode(buf, &fit, &cit, &channel, &q, 1) > 0) {
 		int8_t shift;
@@ -275,29 +361,36 @@ static void sensor_shell_processing_callback(int result, uint8_t *buf, uint32_t
 			continue;
 		}
 
-		int64_t scaled_value = (int64_t)q << shift;
-		bool is_negative = scaled_value < 0;
-		int numerator;
-		int denominator;
-
-		scaled_value = llabs(scaled_value);
-		numerator = (int)FIELD_GET(GENMASK64(31 + shift, 31), scaled_value);
-		denominator =
-			(int)((FIELD_GET(GENMASK64(30, 0), scaled_value) * 1000000) / INT32_MAX);
-
-		if (channel >= ARRAY_SIZE(sensor_channel_name)) {
-			shell_print(ctx->sh, "channel idx=%d value=%s%d.%06d", channel,
-				    is_negative ? "-" : "", numerator, denominator);
-		} else {
-			shell_print(ctx->sh, "channel idx=%d %s value=%s%d.%06d", channel,
-				    sensor_channel_name[channel], is_negative ? "-" : "", numerator,
-				    denominator);
+		if (frame_count > 1) {
+			sample_stats[channel].shift = shift;
+			sample_stats[channel].accumulator += q;
+			sample_stats[channel].count++;
+			sample_stats[channel].is_active = true;
+			continue;
 		}
+
+		sensor_shell_print_q_value(ctx->sh, channel, q, shift);
+	}
+
+	if (frame_count <= 1) {
+		return;
+	}
+
+	shell_print(ctx->sh, "Averages:");
+	for (int i = 0; i < ARRAY_SIZE(sample_stats); ++i) {
+		if (!sample_stats[i].is_active) {
+			continue;
+		}
+
+		int64_t scaled_value = (sample_stats[i].accumulator / sample_stats[i].count);
+
+		sensor_shell_print_q_value(ctx->sh, i, scaled_value, sample_stats[i].shift);
 	}
 }
 
 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 err;
 
@@ -329,8 +422,7 @@ static int cmd_get_sensor(const struct shell *sh, size_t argc, char *argv[])
 				shell_error(sh, "Failed to read channel (%s)", argv[i]);
 				continue;
 			}
-			iodev_sensor_shell_channels[iodev_sensor_shell_read_config.count++] =
-				chan;
+			iodev_sensor_shell_channels[iodev_sensor_shell_read_config.count++] = chan;
 		}
 	}
 
@@ -340,19 +432,31 @@ static int cmd_get_sensor(const struct shell *sh, size_t argc, char *argv[])
 	}
 	iodev_sensor_shell_read_config.sensor = dev;
 
-	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);
+//	sensor_processing_with_callback(&sensor_read_rtio, sensor_shell_processing_callback);
 
 	return 0;
 }
 
+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);
+
 static int cmd_sensor_attr_set(const struct shell *shell_ptr, size_t argc, char *argv[])
 {
 	const struct device *dev;
@@ -462,10 +566,93 @@ static int cmd_sensor_attr_get(const struct shell *shell_ptr, size_t argc, char
 	return 0;
 }
 
-static void channel_name_get(size_t idx, struct shell_static_entry *entry);
+static 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 (dev == NULL) {
+		shell_error(shell_ptr, "Device unknown (%s)", argv[1]);
+		return -ENODEV;
+	}
+
+	if (current_streaming_handle != NULL) {
+		shell_print(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;
+	}
+
+	int op = strcmp("incl", argv[3]) == 0
+			 ? SENSOR_STREAM_DATA_INCLUDE
+			 : (strcmp("drop", argv[3]) == 0
+				    ? SENSOR_STREAM_DATA_DROP
+				    : (strcmp("nop", argv[3]) ? SENSOR_STREAM_DATA_NOP : -1));
+
+	if (op < 0) {
+		shell_error(shell_ptr, "Unknown trigger op (%s)", argv[3]);
+		return -EINVAL;
+	}
+
+	shell_print(shell_ptr, "Enabling stream...");
+	iodev_sensor_shell_stream_config.sensor = dev;
+
+	iodev_sensor_shell_stream_config.count = 1;
+	iodev_sensor_shell_trigger.trigger = SENSOR_TRIG_DATA_READY;
+	iodev_sensor_shell_trigger.opt = op;
+
+	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;
+}
+
+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;
@@ -475,7 +662,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];
@@ -485,27 +672,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++;
@@ -513,6 +719,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);
@@ -549,7 +771,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;
@@ -595,6 +817,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);
@@ -603,8 +837,7 @@ static int cmd_get_sensor_info(const struct shell *sh, size_t argc, char **argv)
 #ifdef CONFIG_SENSOR_INFO
 	const char *null_str = "(null)";
 
-	STRUCT_SECTION_FOREACH(sensor_info, sensor)
-	{
+	STRUCT_SECTION_FOREACH(sensor_info, sensor) {
 		shell_print(sh,
 			    "device name: %s, vendor: %s, model: %s, "
 			    "friendly name: %s",
@@ -685,7 +918,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) {
@@ -701,12 +934,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;
 	}
@@ -739,6 +968,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_ASYNC_API, 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/include/zephyr/drivers/sensor.h b/include/zephyr/drivers/sensor.h
index add8b51306d1c8a..faa3ea3d6944379 100644
--- a/include/zephyr/drivers/sensor.h
+++ b/include/zephyr/drivers/sensor.h
@@ -246,6 +246,9 @@ enum sensor_trigger_type {
 
 	/** Trigger fires when no motion has been detected for a while. */
 	SENSOR_TRIG_STATIONARY,
+
+	SENSOR_TRIG_FIFO_THRESHOLD,
+	SENSOR_TRIG_FIFO_FULL,
 	/**
 	 * Number of all common sensor triggers.
 	 */
@@ -323,6 +326,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.
 	 */
@@ -467,6 +472,15 @@ struct sensor_decoder_api {
 	 */
 	int (*get_timestamp)(const uint8_t *buffer, uint64_t *timestamp_ns);
 
+	/**
+	 * @brief Check if the given trigger type is present
+	 *
+	 * @param[in] buffer The buffer provided on the :c:struct:`rtio` context.
+	 * @param[in] trigger The trigger in question
+	 * @return Whether the tigger is present in the buffer
+	 */
+	bool(*has_trigger)(const uint8_t *buffer, enum sensor_trigger_type trigger);
+
 	/**
 	 * @brief Get the shift count of a particular channel (multiplier)
 	 *
@@ -511,13 +525,38 @@ struct sensor_decoder_api {
 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,
+	/** @brief Flush/clear whatever data is associated with the trigger */
+	SENSOR_STREAM_DATA_DROP,
+	/** @brief Do nothing with the associated trigger data, it may be consumed later */
+	SENSOR_STREAM_DATA_NOP,
+};
+
+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;
 };
@@ -540,12 +579,24 @@ struct sensor_read_config {
 	static enum sensor_channel __channel_array_##name[] = {__VA_ARGS__};                       \
 	static struct sensor_read_config __sensor_read_config_##name = {                           \
 		.sensor = DEVICE_DT_GET(dt_node),                                                  \
+		.is_streaming = false,                                                             \
 		.channels = __channel_array_##name,                                                \
 		.count = ARRAY_SIZE(__channel_array_##name),                                       \
 		.max = ARRAY_SIZE(__channel_array_##name),                                         \
 	};                                                                                         \
 	RTIO_IODEV_DEFINE(name, &__sensor_iodev_api, &__sensor_read_config_##name)
 
+#define SENSOR_DT_STREAM_IODEV(name, dt_node, ...)                                                 \
+	static enum sensor_trigger_type __trigger_array_##name[] = {__VA_ARGS__};                  \
+	static struct sensor_read_config __sensor_read_config_##name = {                           \
+		.sensor = DEVICE_DT_GET(dt_node),                                                  \
+		.is_streaming = true,                                                              \
+		.triggers = __trigger_array_##name,                                                \
+		.count = ARRAY_SIZE(__trigger_array_##name),                                       \
+		.max = ARRAY_SIZE(__trigger_array_##name),                                         \
+	};                                                                                         \
+	RTIO_IODEV_DEFINE(name &__sensor_iodev_api, &__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);
 
@@ -851,7 +902,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;
 	}
 
@@ -859,6 +910,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;
 }
 
@@ -1211,7 +1284,7 @@ struct sensor_info {
  * @param val A pointer to a sensor_value struct.
  * @return The converted value.
  */
-static inline int64_t sensor_value_to_milli(struct sensor_value *val)
+static inline int64_t sensor_value_to_milli(const struct sensor_value *val)
 {
 	return ((int64_t)val->val1 * 1000) + val->val2 / 1000;
 }
@@ -1222,7 +1295,7 @@ static inline int64_t sensor_value_to_milli(struct sensor_value *val)
  * @param val A pointer to a sensor_value struct.
  * @return The converted value.
  */
-static inline int64_t sensor_value_to_micro(struct sensor_value *val)
+static inline int64_t sensor_value_to_micro(const struct sensor_value *val)
 {
 	return ((int64_t)val->val1 * 1000000) + val->val2;
 }
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 000000000000000..aee20b8b10049ce
--- /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