4. 基于TinyML的异常声音检测
任务目标
构建一个可在MCU类设备上运行的异常声音检测系统。系统通过麦克风采集环境声音,提取梅尔频率倒谱系数(MFCC)特征,使用轻量级神经网络(基于TensorFlow Lite for Microcontrollers)对声音进行分类,区分“正常环境声”与“异常声音”(如玻璃破碎、警报声)。最终输出为TFLite模型文件及C++推理代码,可在资源受限的嵌入式设备上完成实时检测。
环境与依赖
- Python 3.8+,用于训练与模型转换
- TensorFlow 2.10+(含TFLite Micro Converter)
- librosa 0.10+(音频特征提取)
- 嵌入式端:TensorFlow Lite for Microcontrollers库(版本≥2.10)
- 硬件:带PDM或I2S麦克风的开发板(如Arduino Nano 33 BLE Sense、ESP32-S3)
实现方案
本案例采用最小可复现闭环:使用Python生成/加载两类音频样本(正常与异常),提取40维MFCC特征(帧长30ms,帧移20ms,取4帧拼接为160维输入向量),训练一个两层全连接分类模型(隐藏层32节点,ReLU激活),转换为TFLite int8量化模型,最后提供C++推理代码框架。由于真实音频数据集体积较大,本案例使用合成音频信号(正弦波组合模拟正常声,方波+噪声模拟异常声)完成全流程验证,用户可替换为真实录音文件。
完整实现代码
# ============ train_anomaly_detector.py ============
import numpy as np
import tensorflow as tf
import librosa
import soundfile as sf
from sklearn.model_selection import train_test_split
# 1. 生成合成音频数据(正常:平滑正弦;异常:方波+突发噪声)
def generate_synthetic_audio(n_samples=200, sr=16000, duration=0.5):
t = np.linspace(0, duration, int(sr*duration), endpoint=False)
normal = []
anomaly = []
for _ in range(n_samples//2):
# 正常声:220Hz+440Hz 正弦叠加,幅度稳定
sig = 0.5*np.sin(2*np.pi*220*t) + 0.3*np.sin(2*np.pi*440*t)
normal.append(sig)
# 异常声:方波基频+随机脉冲噪声
sq = 0.4*np.sign(np.sin(2*np.pi*180*t))
noise = 0.3*np.random.randn(len(t))
pulse = np.zeros_like(t)
pulse[int(0.1*sr):int(0.15*sr)] = 1.0 # 突发脉冲
anomaly.append(sq + noise + pulse)
return np.array(normal), np.array(anomaly)
# 2. 提取MFCC特征(4帧拼接)
def extract_mfcc_features(audio_list, sr=16000):
features = []
for sig in audio_list:
mfcc = librosa.feature.mfcc(y=sig, sr=sr, n_mfcc=40,
n_fft=480, hop_length=320)
# 取前4帧,不足则填充
if mfcc.shape[1] < 4:
pad = np.zeros((40, 4-mfcc.shape[1]))
mfcc = np.hstack([mfcc, pad])
else:
mfcc = mfcc[:, :4]
features.append(mfcc.flatten()) # 160维
return np.array(features)
# 3. 生成数据并划分
normal_audio, anomaly_audio = generate_synthetic_audio()
X_normal = extract_mfcc_features(normal_audio)
X_anomaly = extract_mfcc_features(anomaly_audio)
X = np.vstack([X_normal, X_anomaly])
y = np.array([0]*len(X_normal) + [1]*len(X_anomaly))
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 4. 构建并训练模型
model = tf.keras.Sequential([
tf.keras.layers.Input(shape=(160,)),
tf.keras.layers.Dense(32, activation='relu'),
tf.keras.layers.Dense(2, activation='softmax')
])
model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])
model.fit(X_train, y_train, epochs=30, batch_size=16, validation_split=0.1, verbose=0)
# 5. 转换为TFLite int8量化模型
def representative_dataset():
for i in range(50):
yield [X_train[i].reshape(1, 160).astype(np.float32)]
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type = tf.int8
converter.inference_output_type = tf.int8
tflite_model = converter.convert()
with open('anomaly_detector.tflite', 'wb') as f:
f.write(tflite_model)
# 6. 验证转换后模型精度
interpreter = tf.lite.Interpreter(model_content=tflite_model)
interpreter.allocate_tensors()
input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()
input_scale, input_zero = input_details[0]['quantization']
correct = 0
for i in range(len(X_test)):
x = X_test[i].astype(np.float32)
x_quant = (x / input_scale + input_zero).astype(np.int8)
interpreter.set_tensor(input_details[0]['index'], x_quant.reshape(1, 160))
interpreter.invoke()
out = interpreter.get_tensor(output_details[0]['index'])
pred = np.argmax(out)
if pred == y_test[i]:
correct += 1
print(f"量化模型测试准确率: {correct/len(X_test):.2%}")
嵌入式C++推理代码(关键部分)
// ============ anomaly_detector.cc (Arduino/ESP32) ============
#include <TensorFlowLite.h>
#include "tensorflow/lite/micro/all_ops_resolver.h"
#include "tensorflow/lite/micro/micro_interpreter.h"
#include "anomaly_detector_model.h" // 由tflite文件转换的C数组
// MFCC参数
const int NUM_FRAMES = 4;
const int NUM_MFCC = 40;
const int INPUT_SIZE = NUM_FRAMES * NUM_MFCC; // 160
// 全局变量
tflite::MicroErrorReporter micro_error_reporter;
tflite::AllOpsResolver resolver;
const tflite::Model* model = tflite::GetModel(anomaly_detector_model_tflite);
uint8_t tensor_arena[8 * 1024];
tflite::MicroInterpreter interpreter(model, resolver, tensor_arena, sizeof(tensor_arena));
// 音频采集与MFCC提取(需根据具体硬件实现)
void capture_audio(float* buffer, int len) { /* 硬件相关代码 */ }
void compute_mfcc(const float* audio, int16_t* output) { /* 使用CMSIS-DSP或自实现 */ }
void setup() {
interpreter.AllocateTensors();
}
void loop() {
float audio_buffer[16000 * 0.5]; // 0.5秒@16kHz
capture_audio(audio_buffer, sizeof(audio_buffer)/sizeof(float));
int16_t mfcc_input[INPUT_SIZE];
compute_mfcc(audio_buffer, mfcc_input);
// 输入量化
auto input = interpreter.input(0);
float scale = input->params.scale;
int zero_point = input->params.zero_point;
for (int i = 0; i < INPUT_SIZE; i++) {
input->data.int8[i] = (int8_t)(mfcc_input[i] / scale + zero_point);
}
// 推理
interpreter.Invoke();
// 输出反量化
auto output = interpreter.output(0);
float out_scale = output->params.scale;
int out_zero = output->params.zero_point;
int8_t anomaly_score = output->data.int8[1];
float prob = (anomaly_score - out_zero) * out_scale;
if (prob > 0.7) {
// 触发异常报警
digitalWrite(LED_BUILTIN, HIGH);
} else {
digitalWrite(LED_BUILTIN, LOW);
}
delay(500);
}
运行与验证
- 训练与转换:执行
python train_anomaly_detector.py,脚本将生成anomaly_detector.tflite文件,并打印量化模型在测试集上的准确率。预期输出准确率在95%以上(合成数据线性可分)。 - 模型部署:使用
xxd -i anomaly_detector.tflite > anomaly_detector_model.h将模型转换为C头文件,替换C++代码中的对应引用。 - 硬件验证:将代码烧录至开发板,播放正常环境声音(如风扇声)时LED保持熄灭;播放异常声音(如敲击桌面、金属摩擦声)时LED点亮。若误报频繁,可调整概率阈值(当前0.7)或重新采集真实音频数据训练。
通过标准:量化模型测试准确率≥90%;硬件实测对5次异常声音触发报警≥4次,对5次正常声音误报≤1次。