TinyML：使用 MicroPython 在 ESP32 上进行机器学习

使用 ESP32、加速度计和 MicroPython 近乎实时地从时间序列数据中检测手势。

为什么是这个项目？

我想构建一个使用时间序列数据并可部署到边缘设备（在本例中是 ESP32 微控制器）的TinyML应用程序。我研究了在 ESP32 上使用 MicroPython的机器学习项目，但一无所获（如果我遗漏了什么，请告诉我🙃）。不过，现在越来越多的 C/C++ TinyML 项目将 TensorFlow Lite Micro 与神经网络结合使用。在这个项目的第一个迭代中，我跳过了神经网络，探索了使用标准机器学习算法的可能性。

在进入代码之前，让我们先了解一下基础知识……

TinyML 简介

什么是 TinyML？

TinyML 是机器学习与嵌入式（物联网）设备之间的融合。它赋予高级应用更多“智能”，使其能够利用机器学习驱动。其理念很简单：针对基于规则的逻辑无法满足需求的复杂用例，应用机器学习算法，并在边缘的低功耗设备上运行它们。听起来简单，但执行起来却困难重重。

TinyML 是一个相当新的概念，最早提到它要追溯到 2018 年（？）。关于 TinyML 的定义仍然存在争议。本文中，TinyML 应用是指运行在主频为 MHz 甚至更强大的微控制器（例如 Nvidia Jetson 系列）上的应用程序。Raspberry Pi 也包括在内。TinyML 的其他名称包括 AIoT、边缘分析、边缘人工智能和远边缘计算。请选择您最喜欢的一个。

为什么选择 TinyML？

带宽 - 例如，一台采样率为 100Hz 的设备每小时会产生 36 万个数据点。想象一下，一组这样的设备会产生多少数据。图像和视频的数据量就更加复杂了。
延迟 - “系统接收感知输入并做出响应之间的时间”。在传统的机器学习部署中，数据必须首先发送到机器学习应用程序。这增加了边缘设备等待响应并采取行动的时间。
经济学 ——云计算虽然便宜，但并非绝对便宜。获取大量数据仍然需要投入资金，尤其是在必须实时进行的情况下。
可靠性 ——重新审视带宽示例，在高频采样的情况下，可能很难确保数据按照边缘设备生成的顺序到达目标。
隐私 - TinyML 在设备上处理数据，不会通过网络发送。这减少了数据滥用的可能性。

TinyML 用例

TinyML 的用例范围广泛，从预测性维护到虚拟助手，不一而足。我可能会写一篇文章，探讨 TinyML 的现状、用例以及背后的商业模式。

这个 TinyML 项目是关于什么的？

我着手构建一个 TinyML 系统，该系统可以从时间序列中检测 3 种类型的手势（在本文中我将交替使用手势/动作），存储结果并在网页上将其可视化。

该系统有一个托管在 S3 存储桶上的静态网页、DynamoDB、Golang 微服务，显然还有带有 TinyML 应用程序的边缘设备。

项目架构

虽然这个项目包含更多组件，但本文将介绍机器学习及其在 ESP32 上的部分实现。如果您对完整代码感兴趣，可以在本文末尾找到代码库的链接。

让我们走到边缘去看看硬件。

在边缘

该系统的核心是 ESP32——一款由乐鑫生产的微控制器，主频 240MHz，内置 WiFi+BLE，并支持 MicroPython🐍（我使用的是 MicroPython 1.14）。本项目使用的 IMU 是 MPU6500，具有 6 个自由度 (DoF)——3 个加速度计 (X、Y、Z) 和 3 个角速度计 (X、Y、Z)。此外，还配备了面包板和跳线，用于将所有这些连接起来。

MicroPython

如果您还没有听说过 MicroPython - 它是用于微控制器的 Python。

MicroPython 是 Python 3 编程语言的精简高效实现，包含 Python 标准库的一小部分，并且针对在微控制器和受限环境中运行进行了优化。[链接]

它的性能可能不如 C 或 C++，但它提供了足够的功能，让原型设计变得轻松愉快。尤其适用于对延迟不敏感的物联网应用。（*100Hz 采样率下运行良好）

数据与机器学习

手势定义

您可以在下方找到我收集数据的三个手势。我将它们分别命名为“圆圈”、“X”和“Y”——GIF 也遵循相同的顺序。“圆圈”的含义不言自明。“X”和“Y”是因为手势分别沿着传感器的 X 轴和 Y 轴。理想情况下，我希望在真实的机器数据上检测异常，但这类数据很难获得，也很难复制。而我定义的手势很容易生成，足以测试 MicroPython 和 ESP32 的性能。

实验

机器学习实验分为两部分：第一部分探索时间序列标记对模型性能的影响。我使用了来自传感器的所有可用信号——X、Y、Z 加速度和 X、Y、Z 角速度。此外，我还从推理时间的角度测试了 ESP32 上机器学习的可行性，看看能否实现足够短的推理时间。

第二组实验的重点是模型优化，减少特征空间，选择合适的采样频率，减少错误的推理结果。

收集数据

我使用 Terminal Capture VS Code 扩展程序简化了数据收集。它允许我将 VSC 终端中的传感器数据保存到 txt 文件中，之后我将其转换为 csv 格式。为了打印传感器数据，我编写了以下脚本。它在 ESP32 上启动时运行，采样周期为 10ms（采样率为 100Hz）。10ms 是我能获得一致结果的最低采样周期。我尝试了其他方法，period=5但读数不一致，读数在 5-7ms 之间。这达到了堆栈的第一个限制。尽管如此，10ms（100Hz）已经足够了。

	import utime
	from machine import Timer, Pin, I2C
	from drivers.mpu6500 import MPU6500, SF_DEG_S, SF_M_S2

	# MPU6500 module was adjusted from:
	# https://github.com/tuupola/micropython-mpu9250

	# set up I2C serial communication protcol
	i2c = I2C(scl=Pin(22), sda=Pin(21))

	# create MPU6500 instance
	mpu6500 = MPU6500(i2c, accel_sf=SF_M_S2, gyro_sf=SF_DEG_S)

	# read sensor function
	def read_sensor(timer):
	print(utime.ticks_ms(), mpu6500.acceleration, mpu6500.gyro)

	# hardware timer setup
	timer = Timer(0)
	timer.init(period=10, mode=Timer.PERIODIC, callback=read_sensor)

	import utime
	from machine import Timer, Pin, I2C
	from drivers.mpu6500 import MPU6500, SF_DEG_S, SF_M_S2

	# MPU6500 module was adjusted from:
	# https://github.com/tuupola/micropython-mpu9250

	# set up I2C serial communication protcol
	i2c = I2C(scl=Pin(22), sda=Pin(21))

	# create MPU6500 instance
	mpu6500 = MPU6500(i2c, accel_sf=SF_M_S2, gyro_sf=SF_DEG_S)

	# read sensor function
	def read_sensor(timer):
	print(utime.ticks_ms(), mpu6500.acceleration, mpu6500.gyro)

	# hardware timer setup
	timer = Timer(0)
	timer.init(period=10, mode=Timer.PERIODIC, callback=read_sensor)

view raw data_collect.py hosted with ❤ by GitHub

它的工作原理如下：

标签和标签分发

市面上有很多很棒的标注工具，我使用了labelstud.io来标注我的时间序列数据。在三个定义的手势中，“圆圈”的持续时间最长，大约在800-1000毫秒之间，“X”和“Y”的持续时间在400-600毫秒之间。为了留出缓冲时间，我对这三个标签都使用了1000毫秒的标签跨度。

探索数据集 - EDA

我使用 3D 绘图来查看信号之间是否存在关系。绘制了 1000 毫秒时间跨度内的所有数据点（共 101 个数据点）。

加速度三维图

角速度三维图

很明显，手势加速度和角速度有一个模式。

让我们通过绘制所有信号与其平均值的关系图来再次确认。您可以在相关代码库的 Jupyter Notebook 中找到所有手势和相关矩阵的图表。

注意：信号顶部和底部的截止是由于传感器范围设置为 2G（~19.6 m/s^2）。

机器学习检测手势

由于 ESP32 以及一般的微控制器资源受限，因此我的 TinyML 应用程序有几个要求：

推理时间<<采样周期
ML 模型 < 20kB - 很难将大于 20kB 的文件加载到 ESP32 上（至少使用 MPY 时）

MicroPython 仍是一个年轻的项目，由活跃的社区支持，并且已经开发了许多库。遗憾的是，目前还没有 scikit-learn 或专门针对 MicroPython 的时间序列机器学习库。

如何克服这个问题？

答案是纯 Python 机器学习模型。幸运的是，我找到了一个很棒的库 ( m2cgen )，它可以让你将 scikit-learn 模型导出到 Python、Go、Java（以及许多其他）编程语言。它没有针对时间序列的机器学习模型导出功能。所以，我将使用标准的 scikit-learn 算法。

实际上它看起来像这样：

使用 scikit-learn 在表格数据上训练模型
将 scikit-learn 模型转换为纯 Python 代码
使用纯 Python 模型进行推理

使用 scikit-learn 处理时间序列数据的注意事项

使用 scikit-learn 进行时间序列分析是有代价的——数据必须采用表格格式才能训练模型。有两种方法可以实现这一点 [链接]：

制表（减少）数据

在这种情况下，每个时间点都被视为一个特征，数据在时间上失去了顺序。序列中一个点与前一个点或下一个点之间没有任何依赖关系。
特征提取

在特征提取的情况下，时间序列数据用于计算平均值、最大值、最小值、方差和其他特定于时间序列的变量，然后将其用作模型训练的特征。我们脱离了时间序列领域，在特征领域进行操作。

我选择了数据制表。虽然在 Python 中调用高级库很简单——MicroPython 的数学工具集有限——但我可能无法在 MPY 中提取所有特征。其次，我必须考虑这些特征的计算速度——考虑到有限的资源，计算时间可能比采样周期更长。或许在 TinyML 项目的下一个迭代中可以实现。

机器学习模型训练的数据集变化

标签中的事件位置

这影响了“X”和“Y”手势，因为它们的执行时间在 400-600 毫秒之间。在 1000 毫秒的标签窗口中，可以更改它们的位置。“圆圈”需要 800-1000 毫秒，所以我保留了这个手势的标签。

数据集描述

基线数据集

用于训练和验证的数据集包含收集和标记的动作。
中心 X 和 Y 移动信号

“圆圈”运动占据了整整1000毫秒的时间跨度，无法通过沿时间轴移动来操控。然而，“X”和“Y”的执行时间较短，大约在400-600毫秒之间，因此具有一定的灵活性。我尝试将信号的运动置于1000毫秒窗口的中心，看看模型在这种设置下是否会表现得更好。
中心 X 和 Y 移动信号 + 增强

与之前的情况类似，“X”和“Y”运动位于 1000 毫秒窗口的中心。此外，还引入了一种增强方法。由于运动标记并不“精确”，某些信号的起始位置可能存在偏差。为了弥补这一点，并可能实现更好的泛化，我添加了一个移位——这意味着我使用了一系列小的移位。

对于“X”和“Y”运动，中心位于-20步。对于增强，使用的范围在-20到-15之间。其中一步为10毫秒。

对于“圆圈”，使用的范围在 -2 到 2 之间。

示例：如果原始标签从 0 开始，并且增强数据集移动 -1 步，则增强数据集将以 0-10ms 步长开始。
中心 X 和 Y 移动信号 + SMOTE

类似地，与前两种情况一样，“X”和“Y”居中，但使用了额外的合成过采样（SMOTE），并为训练数据集创建了等量的标签。
窗口末端的 X 和 Y 信号

在这种情况下，“X”和“Y”运动被置于 1000ms 采样窗口的末尾。

数据采样率：
数据采集频率为 100Hz，这允许我进行下采样。您可以在下面找到用于模型训练的频率。

模型评估

在完成初始模型训练、部署和实时数据推理后，我发现 ESP32 上的推理过于敏感——同一个动作会被多次检测。我收集了一个验证数据集，以观察与实时数据结合使用时的情况。每个验证数据集——“圆圈”、“X”、“Y”——都包含 5-6 个手势事件。

我通过滑动推理窗口模拟了实时数据馈送，该窗口在时间序列中滑动时，每一步都会进行一次推理。每个绿色（圆圈）、蓝色（X）和红色（Y）代表一个推理。这些线位于滑动窗口的中心（即 T + 500 毫秒）。

请看下面的示例 - 所有这些模型的准确率都达到 0.95+，但在验证数据集上模拟实时数据时仍然会产生不正确的推理结果。

评估方程式：
我使用了以下方程式。对于每个数据集，我计算了不应该出现的错误标签的比例。我承认我应该标记我的评估数据集，但我需要一种快速的方法来定量评估模型。

标签	方程
圆圈	圆误差 = X+Y / (X+Y+圆)
十	x_error = 圆 + Y / (X + Y + 圆)
是	y_error = 圆 + X / (X + Y + 圆)

基线模型训练结果

最初，我使用了 5 个模型进行基线模型训练，但后来将其从最初的决策树、随机森林、支持向量机、逻辑回归和朴素贝叶斯模型精简为决策树和随机森林。m2cgen 不支持朴素贝叶斯模型，因此我无法将朴素贝叶斯模型转换为纯 Python 代码。逻辑回归和支持向量机在转换为纯 Python 代码时，推理时间存在问题。

随机森林和决策树设置均保留默认设置。

如您所见，关于采样频率和事件在标签中的位置，没有任何结论。此外，我注意到，仅仅通过更改模型的随机种子 (random_seed)，结果就有很大差异。我猜测可以通过收集更多数据来解决这个问题。

您可以在下面看到所有 3 个运动误差的平均值结果。同样，没有明显的优胜者，因此接下来我将使用基线数据集来训练优化模型——坚持基本原则。

测试推理时间

我测试了不同数量估算器的随机森林的推理时间，以查看仍然可用的最高数量。使用 10 个估算器的推理时间约为 4 毫秒，即使在 10 毫秒的采样周期下也是可行的。此外，ESP32 的时钟速度设置为 160MHz，对于实际脚本，我将使用 240MHz（提升 50%），这将进一步缩短推理时间。

注意：随机森林只是决策树的集合 - 如果随机森林通过，决策树也会通过。

优化模型

注意：随机森林只是决策树的集合 - 如果随机森林通过，决策树也会通过。

经过深思熟虑的优化

优化估计器的数量

由于时间限制，估算员的数量必须保持在较低水平 - 理想情况下在 3-5 之间。
优化收集的输入数量

必须为应用程序的不同部分收集 X、Y、Z 轴加速度信号。我考虑创建加速度信号和 1 或 2 个角速度信号的组合。
优化采样率

100Hz 的采样率对于应用来说可能有点过高。而且根据评估结果，它与 50Hz 或 20Hz 的采样率相比没有任何优势。另一方面，10Hz 可能又太慢了。因此，在实验中，我将分别使用 20、25 和 50Hz 的采样率。

为了训练优化模型，我使用了以下参数的网格搜索。

比较结果

您可以在图表中看到所有 3 个手势的结果。蓝点和线表示基线模型（未优化模型），红点和线表示优化模型（网格搜索的结果）。每个图表中的水平线代表所有 3 个手势的平均值。

最佳模型是 ID #2。

将模型 2 与基线模型进行比较

基线模型使用默认设置，以 50Hz 的频率训练所有 6 个信号。这似乎违反直觉，但通过减少信号数量，可以减少错误推断的数量。评估方法与之前所述相同。

虽然通过减少信号数量和调整超参数，推理效果有所提升，但远非完美。推理结果中存在两种现象：首先，存在多组相同的正确推理；其次，存在拖尾推理（主要针对画圈手势）。拖尾推理是由于画圈动作结束时的残留运动造成的。虽然这些残留运动可能被正确分类，但它们是不需要的，必须被过滤掉。理想情况下，每个发送到 REST API 的事件只有一个正确的推理。

推理结果去抖动

我假设模型在“真实”运动中心附近有一个窗口。这意味着，模型会在“真实”运动点前后几毫秒进行推断。此外，还存在不正确的“尾随”推断。

我的去抖动实现基于两个条件。其中一个条件是比较推理缓冲区中第一次推理和最后一次推理之间的时间差（“Circle”、“X”和“Y”推理结果被添加到推理缓冲区中）。另一个条件是评估推理缓冲区中的推理次数。

对于“真实”运动周围的窗口，我假设为 200 毫秒，这实际上允许以 50Hz 采样率进行 9 次推理。因此，推理缓冲区必须至少包含 9 个值。

时间差阈值设置为 450 毫秒。经过实验，它在 50Hz 采样率下效果最佳。它滤除了“画圈”手势的尾随推断，同时仍能检测到“X”和“Y”手势。高于 450 毫秒的值无法检测到这些手势。相反，低于 400 毫秒的时间差阈值会将“尾随”推断归类为单独的手势（通常是错误类型）。

如果满足上述条件 -推理缓冲区中前 9 个元素中最常见的值将作为最终推理结果返回。

注意：这仍处于未完成状态，我正在考虑更智能的重新实施。

防抖实现

	inference_list = []
	DEBOUNCE_THRESHOLD = 9
	TIME_DIFF = 450 #ms

	for st in inference_step: # looping through the whole length of the dataset
	inference = model.score(data)

	if inference in [1,2,3]:
	inference_list.append((time, inference))

	# if there are more or equal than 9 predictions in a list
	# AND
	# the difference between the first and last predictions is > 450 ms

	if len(inference_list) >= DEBOUNCE_THRESHOLD and (inference_list[-1][0] - inference_list[0][0]) >= TIME_DIFF:
	inferences = [x[1] for x in inferences_list[:DEBOUNCE_THRESHOLD]] # gets the first 8 predictions
	inference_final = max(set(inferences), key=inferences.count) # gets the most frequent value
	inference_list = [] # cleans up the list

view raw debounce.py hosted with ❤ by GitHub

	inference_list = []
	DEBOUNCE_THRESHOLD = 9
	TIME_DIFF = 450 #ms

	for st in inference_step: # looping through the whole length of the dataset
	inference = model.score(data)

	if inference in [1,2,3]:
	inference_list.append((time, inference))

	# if there are more or equal than 9 predictions in a list
	# AND
	# the difference between the first and last predictions is > 450 ms

	if len(inference_list) >= DEBOUNCE_THRESHOLD and (inference_list[-1][0] - inference_list[0][0]) >= TIME_DIFF:
	inferences = [x[1] for x in inferences_list[:DEBOUNCE_THRESHOLD]] # gets the first 8 predictions
	inference_final = max(set(inferences), key=inferences.count) # gets the most frequent value
	inference_list = [] # cleans up the list