SmartBench/cmvr-es
8
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add stream audio service

Browse Source
This commit is contained in:
linbo 2026-07-21 17:04:41 +08:00
parent fc041255ab
commit 208b52aef4
17 changed files with 1018 additions and 315 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmvr-es/devices/microphone/abstract_microphone.h
View File

@ -6,6 +6,8 @@
#define CMVR_ES_ABSTRACT_MICROPHONE_H
#pragma once
#include <cstddef>
#include "devices/abstract_device.h"
#include "cmvr/config/microphone_config/microphone_config.pb.h"
namespace cmvr::device{
@ -22,6 +24,12 @@ namespace cmvr::device{
virtual void resume() {}
virtual void setVolume(const int volume) {}
virtual int getVolume() {return 0;}
virtual bool startStreaming() {return true;}
virtual void stopStreaming() {}
virtual void getEncodedFrame(AudioStreamFrameData& frame_data, size_t& index) {}
virtual bool getLatestEncodedFrame(AudioStreamFrameData& frame_data, size_t& next_index) {
return false;
}
protected:
MicrophoneState state_{};

43
cmvr-es/devices/microphone/ffmpeg_microphone/include/ffmpeg_microphone.h
View File

@ -3,26 +3,30 @@
#ifndef CMVR_ES_FFMPEG_MICROPHONE_H
#define CMVR_ES_FFMPEG_MICROPHONE_H
#include <thread>
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <condition_variable>
#include "microphone/abstract_microphone.h"
#include <thread>
#include <boost/lockfree/spsc_queue.hpp>
#include "common/base/ring_buffer.h"
#include "microphone/abstract_microphone.h"
#include "speaker/ffmpeg_speaker/include/ffmpeg_ptr.h"
namespace cmvr::device {
class ffmpegMicroPhone final : public AbstractMicrophone {
public:
// 录制状态
enum class RecordingState {
STOPPED,
RECORDING,
PAUSED
};
public:
ffmpegMicroPhone(const config::FFMpegMicroPhoneConfig& cfg);
explicit ffmpegMicroPhone(const config::FFMpegMicroPhoneConfig& cfg);
~ffmpegMicroPhone() override;
std::string typeName() const override { return "FFMpegMicroPhone"; }
bool init() override;
@ -36,49 +40,52 @@ namespace cmvr::device {
void setVolume(const int volume) override;
int getVolume() override;
bool startStreaming() override;
void stopStreaming() override;
void getEncodedFrame(AudioStreamFrameData& frame_data, size_t& index) override;
bool getLatestEncodedFrame(AudioStreamFrameData& frame_data, size_t& next_index) override;
private:
// 音频采集和编码线程
void audioThread();
// 初始化FFmpeg采集和编码
bool initFFmpeg();
// 关闭FFmpeg资源
void closeFFmpeg();
bool startCapture_(bool write_file);
void stopCapture_();
AudioStreamFormat currentStreamFormat_() const;
void pushEncodedPacket_(const AVPacket* packet);
private:
// FFmpeg采集相关
AVFormatContext* input_fmt_ctx;
AVCodecContext* input_codec_ctx;
AVFrame* input_frame;
// FFmpeg编码相关
AVFormatContext* output_fmt_ctx;
AVOutputFormat* output_fmt;
const AVOutputFormat* output_fmt;
AVStream* audio_st;
AVCodecContext* audio_codec_ctx;
AVCodec* audio_codec;
const AVCodec* audio_codec;
SwrContext* swr_ctx;
AVFrame* audio_frame;
std::string output_file;
std::string format_name;
std::atomic<bool> is_recording;
std::atomic<bool> is_capturing_{false};
std::atomic<bool> write_output_file_{false};
std::atomic<bool> is_paused;
std::mutex pause_mutex;
std::condition_variable pause_cv;
// 音频参数
int sample_rate_;
int channels_;
int64_t next_pts;
private:
std::string input_device_;
std::shared_ptr<std::thread> audio_thread_;
std::mutex mtx_;
std::shared_ptr<SPMCRingBuffer<AudioStreamFrameData>> stream_frame_buffer_;
int stream_count_ = 0;
size_t buffer_size_ = 256;
config::FFMpegMicroPhoneConfig config_;
};

595
cmvr-es/devices/microphone/ffmpeg_microphone/src/ffmpeg_microphone.cpp
View File

@ -1,37 +1,96 @@
#include "common/base/logging/logger.h"
#include "../include/ffmpeg_microphone.h"
#include <algorithm>
#include <chrono>
#include <cstring>
#include <set>
#include <sstream>
using namespace std;
using namespace cmvr::device;
ffmpegMicroPhone::ffmpegMicroPhone(const config::FFMpegMicroPhoneConfig& cfg):input_fmt_ctx(nullptr), input_codec_ctx(nullptr), input_frame(nullptr),
output_fmt_ctx(nullptr), output_fmt(nullptr), audio_st(nullptr),
audio_codec_ctx(nullptr), audio_codec(nullptr), swr_ctx(nullptr),
audio_frame(nullptr), is_recording(false), is_paused(false),
sample_rate_(44100), channels_(2), config_(cfg)
namespace {
std::string avErrorToString(int err)
{
id_ = config_.id();
input_device_ = config_.input_device();
channels_ = config_.channels();
sample_rate_ = config_.samplerate();
state_.volume = config_.volume();
state_.is_initialized = false;
state_.is_running = false;
state_.is_error = false;
state_.error_message= "";
char errbuf[AV_ERROR_MAX_STRING_SIZE] = {0};
av_strerror(err, errbuf, sizeof(errbuf));
return errbuf;
}
ffmpegMicroPhone::~ffmpegMicroPhone() {
void logSupportedAudioInputFormats()
{
static const std::set<std::string> audio_input_names = {
"alsa", "pulse", "oss", "jack", "openal", "sndio", "lavfi", "dshow", "avfoundation"
};
std::ostringstream supported;
void* opaque = nullptr;
const AVInputFormat* fmt = nullptr;
bool first = true;
while ((fmt = av_demuxer_iterate(&opaque)) != nullptr) {
if (!fmt->name || audio_input_names.count(fmt->name) == 0) {
continue;
}
if (!first) {
supported << ", ";
}
supported << fmt->name;
if (fmt->long_name) {
supported << " (" << fmt->long_name << ")";
}
first = false;
}
CMVR_LOG(INFO) << "[ffmpegMicroPhone] Supported audio input formats: "
<< (first ? "none" : supported.str());
}
}
ffmpegMicroPhone::ffmpegMicroPhone(const config::FFMpegMicroPhoneConfig& cfg)
: input_fmt_ctx(nullptr),
input_codec_ctx(nullptr),
input_frame(nullptr),
output_fmt_ctx(nullptr),
output_fmt(nullptr),
audio_st(nullptr),
audio_codec_ctx(nullptr),
audio_codec(nullptr),
swr_ctx(nullptr),
audio_frame(nullptr),
is_paused(false),
sample_rate_(44100),
channels_(2),
next_pts(0),
config_(cfg)
{
id_ = config_.id();
input_device_ = config_.input_device().empty() ? "default" : config_.input_device();
channels_ = config_.channels() > 0 ? config_.channels() : 2;
sample_rate_ = config_.samplerate() > 0 ? config_.samplerate() : 44100;
state_.volume = config_.volume() > 0 ? config_.volume() : 100;
state_.is_initialized = false;
state_.is_running = false;
state_.is_recording = false;
state_.is_error = false;
state_.error_message = "";
stream_frame_buffer_ = std::make_shared<SPMCRingBuffer<AudioStreamFrameData>>(buffer_size_);
}
ffmpegMicroPhone::~ffmpegMicroPhone()
{
stop();
}
bool ffmpegMicroPhone::init() {
bool ffmpegMicroPhone::init()
{
try {
lock_guard lock(mtx_);
stream_frame_buffer_->clear();
state_.is_initialized = true;
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (init): Success, id=" << id_;
return true;
}
catch (const exception& e) {
} catch (const exception& e) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] (init): Failed, id=" << id_ << ": " << e.what();
state_.is_error = true;
state_.error_message = e.what();
@ -39,105 +98,200 @@ bool ffmpegMicroPhone::init() {
}
}
bool ffmpegMicroPhone::start() {
try {
bool ffmpegMicroPhone::start()
{
lock_guard lock(mtx_);
state_.is_initialized = true;
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (start): Success, id=" << id_;
return true;
}
catch (const exception& e) {
CMVR_LOG(ERROR) <<"[ffmpegMicroPhone] (start): Failed, id=" << id_ << ": " << e.what();
state_.is_error = true;
state_.error_message = e.what();
return false;
}
}
bool ffmpegMicroPhone::stop() {
stopRecording();
bool ffmpegMicroPhone::stop()
{
{
lock_guard lock(mtx_);
stream_count_ = 0;
state_.is_recording = false;
}
stopCapture_();
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (stop): Success, id=" << id_;
return true;
}
void ffmpegMicroPhone::pause() {
try {
void ffmpegMicroPhone::pause()
{
is_paused = true;
state_.is_error = false;
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (pause): Success, id=" << id_;
}
catch (const exception& e) {
CMVR_LOG(ERROR) <<"[ffmpegMicroPhone] (pause): Failed, id=" << id_ << ": " << e.what();
state_.is_error = true;
state_.error_message = e.what();
}
}
void ffmpegMicroPhone::resume() {
try {
void ffmpegMicroPhone::resume()
{
is_paused = false;
pause_cv.notify_one();
state_.is_error = false;
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (resume): Success, id=" << id_;
}
catch (const exception& e) {
CMVR_LOG(ERROR) <<"[ffmpegMicroPhone] (resume): Failed, id=" << id_ << ": " << e.what();
state_.is_error = true;
state_.error_message = e.what();
}
void ffmpegMicroPhone::getState(MicrophoneState& state)
{
lock_guard lock(mtx_);
state = state_;
}
void ffmpegMicroPhone::getState(MicrophoneState& state) {
state.is_initialized = state_.is_initialized;
state.is_running = state_.is_running;
state.is_recording = state_.is_recording;
state.is_error = state_.is_error;
state.volume = state_.volume;
state.error_message = state_.error_message;
}
void ffmpegMicroPhone::startRecording(const std::string& outputFilePath) {
try {
if (is_recording) {
void ffmpegMicroPhone::startRecording(const std::string& outputFilePath)
{
lock_guard lock(mtx_);
if (state_.is_recording) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] (startRecording): microphone is already recording";
return;
}
// 从文件路径中提取扩展名
std::string extension;
size_t dotPos = outputFilePath.find_last_of('.');
if (dotPos != std::string::npos) {
extension = outputFilePath.substr(dotPos + 1);
} else {
extension = "wav";
CMVR_LOG(WARNING) << "[ffmpegMicroPhone] (StartRecord): No file extension found, using wav as default, id=" << id_;
if (stream_count_ > 0 || is_capturing_) {
state_.is_error = true;
state_.error_message = "cannot start recording while microphone streaming is active";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] " << state_.error_message;
return;
}
size_t dot_pos = outputFilePath.find_last_of('.');
format_name = dot_pos == std::string::npos ? "wav" : outputFilePath.substr(dot_pos + 1);
output_file = outputFilePath;
format_name = extension;
if (!initFFmpeg()) {
CMVR_LOG(ERROR) <<"[ffmpegMicroPhone] (StartRecord): FFmpeg init error, id=" << id_;
if (!startCapture_(true)) {
state_.is_error = true;
state_.error_message = "FFmpeg init error";
return;
}
is_recording = true;
is_paused = false;
audio_thread_ = std::make_shared<std::thread>(&ffmpegMicroPhone::audioThread, this);
state_.is_recording = true;
}
catch (const exception& e) {
CMVR_LOG(ERROR) <<"[ffmpegMicroPhone] (StartRecord): Failed, id=" << id_ << ": " << e.what();
void ffmpegMicroPhone::stopRecording()
{
{
lock_guard lock(mtx_);
if (!state_.is_recording) {
return;
}
state_.is_recording = false;
}
stopCapture_();
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (stopRecording): Success, id=" << id_;
}
void ffmpegMicroPhone::setVolume(const int volume)
{
lock_guard lock(mtx_);
state_.volume = std::clamp(volume, 0, 100);
}
int ffmpegMicroPhone::getVolume()
{
lock_guard lock(mtx_);
return state_.volume;
}
bool ffmpegMicroPhone::startStreaming()
{
lock_guard lock(mtx_);
if (state_.is_recording) {
state_.is_error = true;
state_.error_message = e.what();
state_.error_message = "cannot start streaming while microphone recording is active";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] " << state_.error_message;
return false;
}
if (stream_count_ == 0) {
stream_frame_buffer_->clear();
format_name = "pcm_s16le";
output_file.clear();
if (!startCapture_(false)) {
state_.is_error = true;
state_.error_message = "FFmpeg init error";
return false;
}
}
++stream_count_;
state_.is_running = true;
state_.is_error = false;
state_.error_message.clear();
return true;
}
void ffmpegMicroPhone::stopStreaming()
{
bool should_stop = false;
{
lock_guard lock(mtx_);
if (stream_count_ > 0) {
--stream_count_;
}
should_stop = stream_count_ == 0 && !state_.is_recording;
}
if (should_stop) {
stopCapture_();
}
}
void ffmpegMicroPhone::stopRecording() {
if (!is_recording)
void ffmpegMicroPhone::getEncodedFrame(AudioStreamFrameData& frame_data, size_t& index)
{
CMVR_LOG(WARNING) << "[ffmpegMicroPhone] (stopRecording): recording has not been started";
if (!stream_frame_buffer_) {
return;
}
auto frame = stream_frame_buffer_->pop(index);
if (frame.has_value()) {
frame_data = frame.value();
}
}
bool ffmpegMicroPhone::getLatestEncodedFrame(AudioStreamFrameData& frame_data, size_t& next_index)
{
if (!stream_frame_buffer_) {
return false;
}
const size_t head = stream_frame_buffer_->getHead();
if (head == 0) {
return false;
}
size_t latest_index = head - 1;
auto frame = stream_frame_buffer_->pop(latest_index);
if (!frame.has_value()) {
return false;
}
frame_data = frame.value();
next_index = latest_index;
return true;
}
bool ffmpegMicroPhone::startCapture_(bool write_file)
{
if (is_capturing_) {
return true;
}
write_output_file_ = write_file;
if (!initFFmpeg()) {
closeFFmpeg();
return false;
}
is_capturing_ = true;
is_paused = false;
state_.is_running = true;
audio_thread_ = std::make_shared<std::thread>(&ffmpegMicroPhone::audioThread, this);
return true;
}
void ffmpegMicroPhone::stopCapture_()
{
if (!is_capturing_) {
return;
}
is_recording = false;
is_capturing_ = false;
is_paused = false;
pause_cv.notify_all();
if (audio_thread_) {
if (audio_thread_->joinable()) {
audio_thread_->join();
@ -145,29 +299,30 @@ void ffmpegMicroPhone::stopRecording() {
audio_thread_.reset();
}
if (output_fmt_ctx) {
if (write_output_file_ && output_fmt_ctx) {
av_write_trailer(output_fmt_ctx);
}
closeFFmpeg();
CMVR_LOG(INFO) << "[ffmpegMicroPhone] (stopRecording): Success, id=" << id_;
write_output_file_ = false;
lock_guard lock(mtx_);
state_.is_running = false;
}
void ffmpegMicroPhone::setVolume(const int volume) {
state_.volume = volume;
}
int ffmpegMicroPhone::getVolume() {
return state_.volume;
}
bool ffmpegMicroPhone::initFFmpeg() {
int err;
bool ffmpegMicroPhone::initFFmpeg()
{
int err = 0;
avdevice_register_all();
// 初始化输入设备
AVInputFormat* input_fmt = av_find_input_format("pulse");
logSupportedAudioInputFormats();
const char* input_format_name = "pulse";
AVInputFormat* input_fmt = av_find_input_format(input_format_name);
if (!input_fmt) {
CMVR_LOG(ERROR) << "Could not find pulse input format";
input_format_name = "alsa";
input_fmt = av_find_input_format(input_format_name);
}
if (!input_fmt) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not find pulse or alsa input format";
return false;
}
@ -176,129 +331,122 @@ bool ffmpegMicroPhone::initFFmpeg() {
av_dict_set(&input_options, "channels", std::to_string(channels_).c_str(), 0);
av_dict_set(&input_options, "format", "s16", 0);
if ((err = avformat_open_input(&input_fmt_ctx, input_device_.c_str(), input_fmt, &input_options)) < 0) {
char errbuf[AV_ERROR_MAX_STRING_SIZE];
av_strerror(err, errbuf, AV_ERROR_MAX_STRING_SIZE);
CMVR_LOG(ERROR) << "Could not open input device: " << errbuf;
CMVR_LOG(INFO) << "[ffmpegMicroPhone] Open input device: format="
<< input_format_name << ", device=" << input_device_;
err = avformat_open_input(&input_fmt_ctx, input_device_.c_str(), input_fmt, &input_options);
av_dict_free(&input_options);
if (err < 0) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not open input device: " << avErrorToString(err);
return false;
}
av_dict_free(&input_options);
if ((err = avformat_find_stream_info(input_fmt_ctx, nullptr)) < 0) {
CMVR_LOG(ERROR) << "Could not find stream info";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not find stream info: " << avErrorToString(err);
return false;
}
// 查找音频流
int audio_stream_index = av_find_best_stream(input_fmt_ctx, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0);
const int audio_stream_index = av_find_best_stream(input_fmt_ctx, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0);
if (audio_stream_index < 0) {
CMVR_LOG(ERROR) << "Could not find audio stream";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not find audio stream";
return false;
}
// 获取输入编解码器上下文
AVCodecParameters* codecpar = input_fmt_ctx->streams[audio_stream_index]->codecpar;
const AVCodec* input_codec = avcodec_find_decoder(codecpar->codec_id);
if (!input_codec) {
CMVR_LOG(ERROR) << "Could not find input codec";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not find input codec";
return false;
}
input_codec_ctx = avcodec_alloc_context3(input_codec);
if (!input_codec_ctx) {
CMVR_LOG(ERROR) << "Could not allocate input codec context";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not allocate input codec context";
return false;
}
if ((err = avcodec_parameters_to_context(input_codec_ctx, codecpar)) < 0) {
CMVR_LOG(ERROR) << "Could not copy input codec parameters";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not copy input codec parameters";
return false;
}
if ((err = avcodec_open2(input_codec_ctx, input_codec, nullptr)) < 0) {
CMVR_LOG(ERROR) << "Could not open input codec";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not open input codec: " << avErrorToString(err);
return false;
}
// 初始化输出格式
if (write_output_file_) {
output_fmt = av_guess_format(format_name.c_str(), nullptr, nullptr);
if (!output_fmt) {
CMVR_LOG(ERROR) << "Unsupported format: " << format_name;
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Unsupported format: " << format_name;
return false;
}
if ((err = avformat_alloc_output_context2(&output_fmt_ctx, output_fmt, nullptr, output_file.c_str())) < 0) {
CMVR_LOG(ERROR) << "Could not allocate output context";
if ((err = avformat_alloc_output_context2(&output_fmt_ctx, nullptr, format_name.c_str(), output_file.c_str())) < 0) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not allocate output context: " << avErrorToString(err);
return false;
}
// 查找音频编码器
output_fmt = output_fmt_ctx->oformat;
audio_codec = avcodec_find_encoder(output_fmt->audio_codec);
} else {
audio_codec = avcodec_find_encoder_by_name("pcm_s16le");
}
if (!audio_codec) {
CMVR_LOG(ERROR) << "Could not find audio encoder";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not find audio encoder";
return false;
}
// 添加音频流
if (write_output_file_) {
audio_st = avformat_new_stream(output_fmt_ctx, nullptr);
if (!audio_st) {
CMVR_LOG(ERROR) << "Could not create audio stream";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not create audio stream";
return false;
}
}
audio_codec_ctx = avcodec_alloc_context3(audio_codec);
if (!audio_codec_ctx) {
CMVR_LOG(ERROR) << "Could not allocate audio codec context";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not allocate audio codec context";
return false;
}
// 设置音频编码参数
audio_codec_ctx->sample_fmt = audio_codec->sample_fmts ? audio_codec->sample_fmts[0] : AV_SAMPLE_FMT_FLTP;
audio_codec_ctx->sample_fmt = audio_codec->sample_fmts ? audio_codec->sample_fmts[0] : AV_SAMPLE_FMT_S16;
audio_codec_ctx->bit_rate = 128000;
audio_codec_ctx->sample_rate = sample_rate_;
audio_codec_ctx->channel_layout = channels_ == 1 ? AV_CH_LAYOUT_MONO : AV_CH_LAYOUT_STEREO;
audio_codec_ctx->channels = channels_;
audio_codec_ctx->time_base = {1, sample_rate_};
if (audio_st) {
audio_st->time_base = {1, sample_rate_};
}
// 打开音频编码器
AVDictionary* opt = nullptr;
if ((err = avcodec_open2(audio_codec_ctx, audio_codec, &opt)) < 0) {
char errbuf[AV_ERROR_MAX_STRING_SIZE];
av_strerror(err, errbuf, AV_ERROR_MAX_STRING_SIZE);
CMVR_LOG(ERROR) << "Could not open audio codec: " << errbuf;
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not open audio codec: " << avErrorToString(err);
av_dict_free(&opt);
return false;
}
av_dict_free(&opt);
// 复制编码参数到流
if (avcodec_parameters_from_context(audio_st->codecpar, audio_codec_ctx) < 0) {
CMVR_LOG(ERROR) << "Could not copy codec parameters to stream";
if (audio_st && avcodec_parameters_from_context(audio_st->codecpar, audio_codec_ctx) < 0) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not copy codec parameters to stream";
return false;
}
// 分配音频帧
input_frame = av_frame_alloc();
audio_frame = av_frame_alloc();
if (!input_frame || !audio_frame) {
CMVR_LOG(ERROR) << "Could not allocate audio frames";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not allocate audio frames";
return false;
}
audio_frame->format = audio_codec_ctx->sample_fmt;
audio_frame->channel_layout = audio_codec_ctx->channel_layout;
audio_frame->channels = audio_codec_ctx->channels;
audio_frame->sample_rate = audio_codec_ctx->sample_rate;
audio_frame->nb_samples = audio_codec_ctx->frame_size ? audio_codec_ctx->frame_size : 1024;
if (av_frame_get_buffer(audio_frame, 0) < 0) {
CMVR_LOG(ERROR) << "Could not allocate audio frame buffers";
return false;
}
// 创建重采样上下文
swr_ctx = swr_alloc();
if (!swr_ctx) {
CMVR_LOG(ERROR) << "Could not allocate resampler context";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not allocate audio frame buffers";
return false;
}
@ -309,25 +457,22 @@ bool ffmpegMicroPhone::initFFmpeg() {
av_get_default_channel_layout(channels_),
input_codec_ctx->sample_fmt,
input_codec_ctx->sample_rate,
0, nullptr);
0,
nullptr);
if (!swr_ctx || swr_init(swr_ctx) < 0) {
CMVR_LOG(ERROR) << "Failed to initialize resampler";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Failed to initialize resampler";
return false;
}
// 打开输出文件
if (!(output_fmt->flags & AVFMT_NOFILE)) {
if (write_output_file_ && output_fmt_ctx && !(output_fmt->flags & AVFMT_NOFILE)) {
if ((err = avio_open(&output_fmt_ctx->pb, output_file.c_str(), AVIO_FLAG_WRITE)) < 0) {
char errbuf[AV_ERROR_MAX_STRING_SIZE];
av_strerror(err, errbuf, AV_ERROR_MAX_STRING_SIZE);
CMVR_LOG(ERROR) << "Could not open output file: " << errbuf;
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not open output file: " << avErrorToString(err);
return false;
}
}
// 写入文件头
if (avformat_write_header(output_fmt_ctx, nullptr) < 0) {
CMVR_LOG(ERROR) << "Could not write output file header";
if (write_output_file_ && output_fmt_ctx && avformat_write_header(output_fmt_ctx, nullptr) < 0) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not write output file header";
return false;
}
@ -335,94 +480,111 @@ bool ffmpegMicroPhone::initFFmpeg() {
return true;
}
void ffmpegMicroPhone::audioThread() {
void ffmpegMicroPhone::audioThread()
{
AVPacket* input_pkt = av_packet_alloc();
AVPacket* output_pkt = av_packet_alloc();
if (!input_pkt || !output_pkt) {
CMVR_LOG(ERROR) << "Could not allocate packets";
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not allocate packets";
av_packet_free(&input_pkt);
av_packet_free(&output_pkt);
return;
}
while (is_recording) {
while (is_capturing_) {
if (is_paused) {
std::unique_lock<std::mutex> lock(pause_mutex);
pause_cv.wait(lock, [this] { return !is_paused || !is_recording; });
if (!is_recording) break;
unique_lock<std::mutex> lock(pause_mutex);
pause_cv.wait(lock, [this] { return !is_paused || !is_capturing_; });
continue;
}
// 从输入设备读取数据
int ret = av_read_frame(input_fmt_ctx, input_pkt);
if (ret < 0) {
if (ret == AVERROR(EAGAIN)) continue;
const int read_ret = av_read_frame(input_fmt_ctx, input_pkt);
if (read_ret < 0) {
if (read_ret == AVERROR(EAGAIN)) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
continue;
}
break;
}
// 解码音频帧
ret = avcodec_send_packet(input_codec_ctx, input_pkt);
if (ret < 0) {
int ret = avcodec_send_packet(input_codec_ctx, input_pkt);
av_packet_unref(input_pkt);
if (ret < 0) {
continue;
}
while (avcodec_receive_frame(input_codec_ctx, input_frame) == 0) {
if (input_frame->format != AV_SAMPLE_FMT_S16) {
CMVR_LOG(ERROR) << "Input frame format mismatch";
return;
}
// 重采样
int dst_nb_samples = av_rescale_rnd(
const int dst_nb_samples = static_cast<int>(av_rescale_rnd(
swr_get_delay(swr_ctx, sample_rate_) + input_frame->nb_samples,
sample_rate_, sample_rate_, AV_ROUND_UP
);
sample_rate_,
sample_rate_,
AV_ROUND_UP));
int ret = swr_convert(
swr_ctx, audio_frame->data, dst_nb_samples, // 使用计算出的样本数
(const uint8_t**)input_frame->data, input_frame->nb_samples
);
if (ret < 0) {
CMVR_LOG(ERROR) << "swr_convert failed";
av_frame_unref(input_frame);
if (dst_nb_samples > audio_frame->nb_samples) {
av_frame_unref(audio_frame);
audio_frame->format = audio_codec_ctx->sample_fmt;
audio_frame->channel_layout = audio_codec_ctx->channel_layout;
audio_frame->channels = audio_codec_ctx->channels;
audio_frame->sample_rate = audio_codec_ctx->sample_rate;
audio_frame->nb_samples = dst_nb_samples;
const int buffer_ret = av_frame_get_buffer(audio_frame, 0);
if (buffer_ret < 0) {
CMVR_LOG(ERROR) << "[ffmpegMicroPhone] Could not grow audio frame buffer: "
<< avErrorToString(buffer_ret);
break;
}
}
if (av_frame_make_writable(audio_frame) < 0) {
continue;
}
audio_frame->nb_samples = ret; // 更新实际转换的样本数
// 更准确的pts计算
audio_frame->pts = av_rescale_q(
next_pts, (AVRational){1, sample_rate_}, audio_codec_ctx->time_base
);
next_pts += audio_frame->nb_samples; // 使用实际样本数更新
ret = swr_convert(
swr_ctx,
audio_frame->data,
dst_nb_samples,
const_cast<const uint8_t**>(input_frame->data),
input_frame->nb_samples);
av_frame_unref(input_frame);
if (ret < 0) {
continue;
}
audio_frame->nb_samples = ret;
audio_frame->pts = av_rescale_q(next_pts, AVRational{1, sample_rate_}, audio_codec_ctx->time_base);
next_pts += audio_frame->nb_samples;
// 编码音频帧
ret = avcodec_send_frame(audio_codec_ctx, audio_frame);
if (ret < 0) {
continue;
}
while (avcodec_receive_packet(audio_codec_ctx, output_pkt) == 0) {
// 写入数据包
if (write_output_file_ && output_fmt_ctx && audio_st) {
av_packet_rescale_ts(output_pkt, audio_codec_ctx->time_base, audio_st->time_base);
output_pkt->stream_index = audio_st->index;
if (av_interleaved_write_frame(output_fmt_ctx, output_pkt) < 0) {
av_packet_unref(output_pkt);
break;
}
}
pushEncodedPacket_(output_pkt);
av_packet_unref(output_pkt);
}
}
av_packet_unref(input_pkt);
}
// 刷新编码器
avcodec_send_frame(audio_codec_ctx, nullptr);
while (avcodec_receive_packet(audio_codec_ctx, output_pkt) == 0) {
if (write_output_file_ && output_fmt_ctx && audio_st) {
av_packet_rescale_ts(output_pkt, audio_codec_ctx->time_base, audio_st->time_base);
output_pkt->stream_index = audio_st->index;
if (av_interleaved_write_frame(output_fmt_ctx, output_pkt) < 0) {
av_packet_unref(output_pkt);
break;
}
}
pushEncodedPacket_(output_pkt);
av_packet_unref(output_pkt);
}
@ -430,38 +592,71 @@ void ffmpegMicroPhone::audioThread() {
av_packet_free(&output_pkt);
}
void ffmpegMicroPhone::closeFFmpeg() {
AudioStreamFormat ffmpegMicroPhone::currentStreamFormat_() const
{
if (!audio_codec_ctx) {
return AudioStreamFormat::UNKNOWN;
}
switch (audio_codec_ctx->codec_id) {
case AV_CODEC_ID_AAC:
return AudioStreamFormat::AAC;
case AV_CODEC_ID_MP3:
return AudioStreamFormat::MP3;
case AV_CODEC_ID_PCM_S16LE:
case AV_CODEC_ID_PCM_S16BE:
return AudioStreamFormat::PCM;
default:
return AudioStreamFormat::UNKNOWN;
}
}
void ffmpegMicroPhone::pushEncodedPacket_(const AVPacket* packet)
{
if (!packet || !packet->data || packet->size <= 0 || stream_count_ <= 0 || !stream_frame_buffer_) {
return;
}
AudioStreamFrameData frame;
frame.data.assign(packet->data, packet->data + packet->size);
frame.sample_rate = sample_rate_;
frame.channels = channels_;
frame.format = currentStreamFormat_();
frame.codec = audio_codec_ctx && audio_codec_ctx->codec ? audio_codec_ctx->codec->name : "pcm_s16le";
frame.pts = packet->pts;
frame.nb_samples = audio_frame ? audio_frame->nb_samples : 0;
stream_frame_buffer_->push(frame);
}
void ffmpegMicroPhone::closeFFmpeg()
{
if (input_frame) {
av_frame_unref(input_frame); // 释放内部缓冲区（如果有）
av_frame_free(&input_frame); // 释放 AVFrame 结构体
av_frame_free(&input_frame);
input_frame = nullptr;
}
if (audio_frame) {
av_frame_unref(audio_frame); // 释放内部缓冲区（如果有）
av_frame_free(&audio_frame); // 释放 AVFrame 结构体
av_frame_free(&audio_frame);
audio_frame = nullptr;
}
if (swr_ctx) swr_free(&swr_ctx);
if (swr_ctx) {
swr_free(&swr_ctx);
}
if (input_codec_ctx) {
avcodec_close(input_codec_ctx);
avcodec_free_context(&input_codec_ctx);
}
if (audio_codec_ctx) {
avcodec_close(audio_codec_ctx);
avcodec_free_context(&audio_codec_ctx);
}
if (input_fmt_ctx) {
avformat_close_input(&input_fmt_ctx);
}
if (output_fmt_ctx) {
if (!(output_fmt->flags & AVFMT_NOFILE)) {
if (output_fmt && !(output_fmt->flags & AVFMT_NOFILE)) {
avio_closep(&output_fmt_ctx->pb);
}
avformat_free_context(output_fmt_ctx);
output_fmt_ctx = nullptr;
}
output_fmt = nullptr;
audio_st = nullptr;
audio_codec = nullptr;
}

2
cmvr-es/devices/speaker/abstract_speaker.h
View File

@ -21,6 +21,8 @@ namespace cmvr::device{
virtual int getVolume() const {return 0;}
virtual void pause() {}
virtual void resume() {}
virtual bool pushAudioFrame(const AudioStreamFrameData& frame_data) { return false; }
virtual void stopStreaming() {}
protected:
SpeakerState state_{};

17
cmvr-es/devices/speaker/ffmpeg_speaker/include/ffmpeg_speaker.h
View File

@ -6,6 +6,7 @@
#include <thread>
#include <mutex>
#include <condition_variable>
#include <string>
#include "speaker/abstract_speaker.h"
#include <boost/lockfree/spsc_queue.hpp>
#include <pulse/simple.h>
@ -33,6 +34,8 @@ namespace cmvr::device {
void pause() override;
void resume() override;
void getState(SpeakerState& state) override;
bool pushAudioFrame(const AudioStreamFrameData& frame_data) override;
void stopStreaming() override;
void resetPlayState();
bool initPulseDevice_();
@ -40,6 +43,11 @@ namespace cmvr::device {
private:
void decode_audio_();
void play_audio_();
bool startStreamingPlayback_(const AudioStreamFrameData& frame_data);
bool pushPcmFrame_(const AudioStreamFrameData& frame_data);
bool decodeStreamFrame_(const AudioStreamFrameData& frame_data);
bool initStreamDecoder_(const AudioStreamFrameData& frame_data);
void releaseStreamDecoder_();
pa_simple* pulse_simple_ = nullptr; // 修改：PulseAudio 简单 API 句柄
pa_sample_spec sample_spec_{}; // 新增：PulseAudio 采样规格
@ -64,6 +72,15 @@ namespace cmvr::device {
// 音频时钟同步
std::atomic<int64_t> audio_pts_{0};
std::chrono::time_point<std::chrono::steady_clock> playback_start_time_;
std::mutex stream_decode_mtx_;
AVCodecContext* stream_decoder_ctx_ = nullptr;
SwrContext* stream_swr_ctx_ = nullptr;
AVFrame* stream_decode_frame_ = nullptr;
AVPacket* stream_decode_packet_ = nullptr;
std::string stream_codec_;
int stream_input_sample_rate_ = 0;
int stream_input_channels_ = 0;
bool is_streaming_input_ = false;
config::FFMpegSpeakerConfig config_;
};

289
cmvr-es/devices/speaker/ffmpeg_speaker/src/ffmpeg_speaker.cpp
View File

@ -4,6 +4,9 @@
//
#include <filesystem>
#include <algorithm>
#include <climits>
#include <cstring>
#include "../include/ffmpeg_speaker.h"
@ -18,7 +21,7 @@ ffmpegSpeaker::ffmpegSpeaker(const config::FFMpegSpeakerConfig& cfg):config_(cfg
try {
id_ = config_.id();
memset(&sample_spec_, 0, sizeof(sample_spec_));
// state_.volume = config_.volume();
state_.volume = 100;
}
catch (const exception& e) {
CMVR_LOG(ERROR) << "[ffmpegSpeaker] ([ffmpegSpeaker]): Failed to parse config: " << e.what();
@ -76,6 +79,8 @@ bool ffmpegSpeaker::start() {
void ffmpegSpeaker::resetPlayState()
{
releaseStreamDecoder_();
if (pulse_simple_) {
pa_simple_flush(pulse_simple_, nullptr);
pa_simple_free(pulse_simple_);
@ -89,6 +94,7 @@ void ffmpegSpeaker::resetPlayState()
}
state_.is_initialized = false;
is_streaming_input_ = false;
audio_path_.clear();
CMVR_LOG(INFO) << "[ffmpegSpeaker] (resetPlayState): Success, id=" << id_;
@ -150,6 +156,7 @@ void ffmpegSpeaker::resume() {
return;
}
state_.is_paused = false;
cv_pause_.notify_all();
CMVR_LOG(INFO) << "[ffmpegSpeaker] (resume): Success, id=" << id_;
}
catch (const exception& e) {
@ -531,6 +538,286 @@ void ffmpegSpeaker::play_audio_() {
CMVR_LOG(INFO) << "[ffmpegSpeaker] Playback finished";
}
bool ffmpegSpeaker::pushAudioFrame(const AudioStreamFrameData& frame_data)
{
if (frame_data.data.empty()) {
return true;
}
bool need_start = false;
bool need_restart = false;
{
std::lock_guard<std::mutex> lock(mtx_);
need_start = !state_.is_running;
need_restart = state_.is_running && is_streaming_input_ &&
(frame_data.sample_rate != sample_rate_ || frame_data.channels != channels_);
if (state_.is_running && !is_streaming_input_) {
CMVR_LOG(ERROR) << "[ffmpegSpeaker] cannot stream audio while file playback is active";
return false;
}
}
if (need_restart) {
stopStreaming();
need_start = true;
}
if (need_start && !startStreamingPlayback_(frame_data)) {
return false;
}
if (frame_data.format == AudioStreamFormat::PCM ||
frame_data.codec == "pcm_s16le" ||
frame_data.codec == "pcm") {
return pushPcmFrame_(frame_data);
}
return decodeStreamFrame_(frame_data);
}
void ffmpegSpeaker::stopStreaming()
{
{
std::lock_guard<std::mutex> lock(mtx_);
if (!is_streaming_input_) {
return;
}
state_.is_decoding = false;
state_.is_paused = false;
is_streaming_input_ = false;
}
cv_pause_.notify_all();
if (play_thread_ && play_thread_->joinable()) {
play_thread_->join();
play_thread_.reset();
}
resetPlayState();
}
bool ffmpegSpeaker::startStreamingPlayback_(const AudioStreamFrameData& frame_data)
{
std::lock_guard<std::mutex> lock(mtx_);
sample_rate_ = frame_data.sample_rate > 0 ? frame_data.sample_rate : 44100;
channels_ = frame_data.channels > 0 ? frame_data.channels : 2;
AudioFrame queued_frame;
while (audio_queue_.pop(queued_frame)) {
}
if (!initPulseDevice_()) {
CMVR_LOG(ERROR) << "[ffmpegSpeaker] PulseAudio device initialization failed";
return false;
}
state_.is_running = true;
state_.is_decoding = true;
state_.is_paused = false;
is_streaming_input_ = true;
play_thread_ = std::make_shared<std::thread>([this]() {
try {
this->play_audio_();
} catch (const std::exception& e) {
CMVR_LOG(ERROR) << "[ffmpegSpeaker] Stream play thread exception: " << e.what();
std::lock_guard<std::mutex> lock(mtx_);
state_.is_running = false;
state_.is_decoding = false;
}
});
return true;
}
bool ffmpegSpeaker::pushPcmFrame_(const AudioStreamFrameData& frame_data)
{
const size_t sample_count = frame_data.data.size() / sizeof(int16_t);
if (sample_count == 0) {
return true;
}
auto buffer = std::make_shared<std::vector<int16_t>>(sample_count);
std::memcpy(buffer->data(), frame_data.data.data(), sample_count * sizeof(int16_t));
float volume_scale = 1.0f;
{
std::lock_guard<std::mutex> lock(mtx_);
volume_scale = static_cast<float>(state_.volume) / 100.0f;
}
for (int16_t& sample : *buffer) {
const float scaled = static_cast<float>(sample) * volume_scale;
sample = static_cast<int16_t>(std::clamp(scaled, -32768.f, 32767.f));
}
while (!audio_queue_.push(buffer)) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
std::lock_guard<std::mutex> lock(mtx_);
if (!state_.is_running) {
return false;
}
}
return true;
}
bool ffmpegSpeaker::decodeStreamFrame_(const AudioStreamFrameData& frame_data)
{
std::lock_guard<std::mutex> decode_lock(stream_decode_mtx_);
if (!stream_decoder_ctx_ ||
stream_codec_ != frame_data.codec ||
stream_input_sample_rate_ != frame_data.sample_rate ||
stream_input_channels_ != frame_data.channels) {
releaseStreamDecoder_();
if (!initStreamDecoder_(frame_data)) {
return false;
}
}
av_packet_unref(stream_decode_packet_);
if (av_new_packet(stream_decode_packet_, static_cast<int>(frame_data.data.size())) < 0) {
return false;
}
std::memcpy(stream_decode_packet_->data, frame_data.data.data(), frame_data.data.size());
stream_decode_packet_->pts = frame_data.pts;
int ret = avcodec_send_packet(stream_decoder_ctx_, stream_decode_packet_);
av_packet_unref(stream_decode_packet_);
if (ret < 0) {
return false;
}
while ((ret = avcodec_receive_frame(stream_decoder_ctx_, stream_decode_frame_)) == 0) {
if (!stream_swr_ctx_) {
stream_swr_ctx_ = swr_alloc_set_opts(nullptr,
av_get_default_channel_layout(channels_),
AV_SAMPLE_FMT_S16,
sample_rate_,
av_get_default_channel_layout(stream_decode_frame_->channels),
static_cast<AVSampleFormat>(stream_decode_frame_->format),
stream_decode_frame_->sample_rate,
0,
nullptr);
if (!stream_swr_ctx_ || swr_init(stream_swr_ctx_) < 0) {
CMVR_LOG(ERROR) << "[ffmpegSpeaker] Failed to init stream resampler";
av_frame_unref(stream_decode_frame_);
return false;
}
}
const int64_t max_samples64 = av_rescale_rnd(
swr_get_delay(stream_swr_ctx_, stream_decode_frame_->sample_rate) + stream_decode_frame_->nb_samples,
sample_rate_,
stream_decode_frame_->sample_rate,
AV_ROUND_UP);
const int max_samples = static_cast<int>(std::min<int64_t>(max_samples64, INT_MAX));
auto buffer = std::make_shared<std::vector<int16_t>>(max_samples * channels_);
uint8_t* out[] = {reinterpret_cast<uint8_t*>(buffer->data()), nullptr};
const int out_samples = swr_convert(stream_swr_ctx_,
out,
max_samples,
const_cast<const uint8_t**>(stream_decode_frame_->data),
stream_decode_frame_->nb_samples);
av_frame_unref(stream_decode_frame_);
if (out_samples <= 0) {
continue;
}
buffer->resize(out_samples * channels_);
float volume_scale = 1.0f;
{
std::lock_guard<std::mutex> lock(mtx_);
volume_scale = static_cast<float>(state_.volume) / 100.0f;
}
for (int16_t& sample : *buffer) {
const float scaled = static_cast<float>(sample) * volume_scale;
sample = static_cast<int16_t>(std::clamp(scaled, -32768.f, 32767.f));
}
while (!audio_queue_.push(buffer)) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
std::lock_guard<std::mutex> lock(mtx_);
if (!state_.is_running) {
return false;
}
}
}
return ret == AVERROR(EAGAIN) || ret == AVERROR_EOF;
}
bool ffmpegSpeaker::initStreamDecoder_(const AudioStreamFrameData& frame_data)
{
const AVCodec* decoder = nullptr;
if (!frame_data.codec.empty()) {
decoder = avcodec_find_decoder_by_name(frame_data.codec.c_str());
}
if (!decoder) {
AVCodecID codec_id = AV_CODEC_ID_NONE;
if (frame_data.format == AudioStreamFormat::AAC) {
codec_id = AV_CODEC_ID_AAC;
} else if (frame_data.format == AudioStreamFormat::MP3) {
codec_id = AV_CODEC_ID_MP3;
}
if (codec_id != AV_CODEC_ID_NONE) {
decoder = avcodec_find_decoder(codec_id);
}
}
if (!decoder) {
CMVR_LOG(ERROR) << "[ffmpegSpeaker] Unsupported stream audio codec: " << frame_data.codec;
return false;
}
stream_decoder_ctx_ = avcodec_alloc_context3(decoder);
if (!stream_decoder_ctx_) {
return false;
}
stream_decoder_ctx_->sample_rate = frame_data.sample_rate > 0 ? frame_data.sample_rate : sample_rate_;
stream_decoder_ctx_->channels = frame_data.channels > 0 ? frame_data.channels : channels_;
stream_decoder_ctx_->channel_layout =
stream_decoder_ctx_->channels == 1 ? AV_CH_LAYOUT_MONO : AV_CH_LAYOUT_STEREO;
if (avcodec_open2(stream_decoder_ctx_, decoder, nullptr) < 0) {
releaseStreamDecoder_();
return false;
}
stream_decode_frame_ = av_frame_alloc();
stream_decode_packet_ = av_packet_alloc();
if (!stream_decode_frame_ || !stream_decode_packet_) {
releaseStreamDecoder_();
return false;
}
stream_codec_ = frame_data.codec;
stream_input_sample_rate_ = frame_data.sample_rate;
stream_input_channels_ = frame_data.channels;
return true;
}
void ffmpegSpeaker::releaseStreamDecoder_()
{
if (stream_swr_ctx_) {
swr_free(&stream_swr_ctx_);
}
if (stream_decode_frame_) {
av_frame_free(&stream_decode_frame_);
}
if (stream_decode_packet_) {
av_packet_free(&stream_decode_packet_);
}
if (stream_decoder_ctx_) {
avcodec_free_context(&stream_decoder_ctx_);
}
stream_codec_.clear();
stream_input_sample_rate_ = 0;
stream_input_channels_ = 0;
}
bool ffmpegSpeaker::initPulseDevice_() {
// 验证参数
if (sample_rate_ <= 0 || channels_ <= 0) {

19
cmvr-es/devices/state_define.h
View File

@ -6,6 +6,7 @@
#define CMVR_ES_STATE_DEFINE_H
#include <atomic>
#include <cstdint>
#include <cmath>
#include <iostream>
#include <vector>
@ -39,6 +40,24 @@ namespace cmvr::device{
} AGVState;
enum class AudioStreamFormat {
PCM = 0,
MP3 = 1,
AAC = 2,
WAV = 3,
UNKNOWN = 99
};
struct AudioStreamFrameData {
std::vector<uint8_t> data;
int sample_rate = 44100;
int channels = 2;
AudioStreamFormat format = AudioStreamFormat::PCM;
std::string codec = "pcm_s16le";
int64_t pts = 0;
int nb_samples = 0;
};
// ------------------------------------- robot -------------------------------------
typedef enum {
FORWARD, BACKWARD,

1
cmvr-es/service/grpc/include/grpc_microphone_service.h
View File

@ -22,6 +22,7 @@ namespace cmvr::service
grpc::Status StopRecord(grpc::ServerContext* context, const api::StopMicRecordingCommand_Request* request,api::StopMicRecordingCommand_Feedback* response) override;
grpc::Status PauseRecord(grpc::ServerContext* context, const api::PauseMicRecordingCommand_Request* request,api::PauseMicRecordingCommand_Feedback* response) override;
grpc::Status ResumeRecord(grpc::ServerContext* context, const api::ResumeMicRecordingCommand_Request* request,api::ResumeMicRecordingCommand_Feedback* response) override;
grpc::Status StreamAudio(grpc::ServerContext* context, const api::StreamMicAudioCommand_Request* request, grpc::ServerWriter<api::StreamMicAudioCommand_Feedback>* writer) override;
grpc::Status SetVolume(grpc::ServerContext* context, const api::SetMicPhoneVolumeCommand_Request* request,api::SetMicPhoneVolumeCommand_Feedback* response) override;
grpc::Status GetVolume(grpc::ServerContext* context, const api::GetMicPhoneVolumeCommand_Request* request,api::GetMicPhoneVolumeCommand_Feedback* response) override;
private:

1
cmvr-es/service/grpc/include/grpc_speaker_service.h
View File

@ -17,6 +17,7 @@ namespace cmvr::service {
~gRPCSpeakerServiceImpl() override = default;
grpc::Status GetStatus(grpc::ServerContext* context, const api::GetSpeakerStateCommand_Request* request,api::GetSpeakerStateCommand_Feedback* response) override;
grpc::Status PlayAudio(grpc::ServerContext* context, const api::PlayAudioCommand_Request* request,api::PlayAudioCommand_Feedback* response) override;
grpc::Status StreamAudio(grpc::ServerContext* context, grpc::ServerReader<api::StreamSpeakerAudioCommand_Request>* reader, api::StreamSpeakerAudioCommand_Feedback* response) override;
grpc::Status StopPlayback(grpc::ServerContext* context, const api::StopSpeakerCommand_Request* request,api::StopSpeakerCommand_Feedback* response) override;
grpc::Status PausePlayback(grpc::ServerContext* context, const api::PauseSpeakerCommand_Request* request,api::PauseSpeakerCommand_Feedback* response) override;
grpc::Status ResumePlayback(grpc::ServerContext* context, const api::ResumeSpeakerCommand_Request* request,api::ResumeSpeakerCommand_Feedback* response) override;

82
cmvr-es/service/grpc/src/grpc_microphone_service.cpp
View File

@ -1,4 +1,6 @@
#include "common/base/logging/logger.h"
#include <chrono>
#include <thread>
//
// Created by linbo on 2025/6/13.
// Created by xtkuang on 2025/6/13.
@ -18,6 +20,31 @@ grpc::Status failResponse(ResponseT* response, const std::string& message) {
setCurrentTimestamp(response->mutable_header()->mutable_timestamp());
return grpc::Status::OK;
}
cmvr::api::AudioData_AudioFormat toProtoAudioFormat(AudioStreamFormat format) {
switch (format) {
case AudioStreamFormat::PCM:
return cmvr::api::AudioData_AudioFormat_PCM;
case AudioStreamFormat::MP3:
return cmvr::api::AudioData_AudioFormat_MP3;
case AudioStreamFormat::AAC:
return cmvr::api::AudioData_AudioFormat_AAC;
case AudioStreamFormat::WAV:
return cmvr::api::AudioData_AudioFormat_WAV;
default:
return cmvr::api::AudioData_AudioFormat_PCM;
}
}
void fillProtoAudioData(const AudioStreamFrameData& frame, cmvr::api::AudioData* audio) {
audio->set_data(frame.data.data(), frame.data.size());
audio->set_sample_rate(frame.sample_rate);
audio->set_channels(frame.channels);
audio->set_format(toProtoAudioFormat(frame.format));
audio->set_codec(frame.codec);
audio->set_pts(frame.pts);
audio->set_nb_samples(frame.nb_samples);
}
}
gRPCMicroPhoneServiceImpl::gRPCMicroPhoneServiceImpl(): dmgr_(DeviceManager::getInstance()) {}
@ -151,6 +178,61 @@ grpc::Status gRPCMicroPhoneServiceImpl::ResumeRecord(grpc::ServerContext* contex
}
}
grpc::Status gRPCMicroPhoneServiceImpl::StreamAudio(grpc::ServerContext* context,
const api::StreamMicAudioCommand_Request* request,
grpc::ServerWriter<api::StreamMicAudioCommand_Feedback>* writer) {
const string dev_id = request->header().device_id();
CMVR_LOG(INFO) << "[gRPCMicroPhoneServiceImpl] (StreamAudio): id=" << dev_id;
const auto dev = dmgr_.getDevice<AbstractMicrophone>(dev_id);
if (!dev) {
api::StreamMicAudioCommand_Feedback feedback;
feedback.mutable_header()->set_success(false);
feedback.mutable_header()->set_error_message("Microphone device not found: " + dev_id);
setCurrentTimestamp(feedback.mutable_header()->mutable_timestamp());
writer->Write(feedback);
return grpc::Status::OK;
}
if (!dev->start()) {
api::StreamMicAudioCommand_Feedback feedback;
feedback.mutable_header()->set_success(false);
feedback.mutable_header()->set_error_message("Failed to start microphone: " + dev_id);
setCurrentTimestamp(feedback.mutable_header()->mutable_timestamp());
writer->Write(feedback);
return grpc::Status::OK;
}
if (!dev->startStreaming()) {
api::StreamMicAudioCommand_Feedback feedback;
feedback.mutable_header()->set_success(false);
feedback.mutable_header()->set_error_message("Failed to start microphone streaming: " + dev_id);
setCurrentTimestamp(feedback.mutable_header()->mutable_timestamp());
writer->Write(feedback);
return grpc::Status::OK;
}
size_t index = 0;
while (!context->IsCancelled()) {
AudioStreamFrameData frame;
dev->getEncodedFrame(frame, index);
if (frame.data.empty()) {
std::this_thread::sleep_for(std::chrono::milliseconds(2));
continue;
}
api::StreamMicAudioCommand_Feedback feedback;
feedback.mutable_header()->set_success(true);
setCurrentTimestamp(feedback.mutable_header()->mutable_timestamp());
fillProtoAudioData(frame, feedback.mutable_audio());
if (!writer->Write(feedback)) {
break;
}
}
dev->stopStreaming();
return grpc::Status::OK;
}
grpc::Status gRPCMicroPhoneServiceImpl::SetVolume(grpc::ServerContext* context,
const api::SetMicPhoneVolumeCommand_Request* request, api::SetMicPhoneVolumeCommand_Feedback* response) {
try {

70
cmvr-es/service/grpc/src/grpc_speaker_service.cpp
View File

@ -1,4 +1,5 @@
#include "common/base/logging/logger.h"
#include <memory>
//
// Created by xtkuang on 2025/6/10.
//
@ -18,6 +19,33 @@ grpc::Status failResponse(ResponseT* response, const std::string& message) {
setCurrentTimestamp(response->mutable_header()->mutable_timestamp());
return grpc::Status::OK;
}
AudioStreamFormat fromProtoAudioFormat(cmvr::api::AudioData_AudioFormat format) {
switch (format) {
case cmvr::api::AudioData_AudioFormat_PCM:
return AudioStreamFormat::PCM;
case cmvr::api::AudioData_AudioFormat_MP3:
return AudioStreamFormat::MP3;
case cmvr::api::AudioData_AudioFormat_AAC:
return AudioStreamFormat::AAC;
case cmvr::api::AudioData_AudioFormat_WAV:
return AudioStreamFormat::WAV;
default:
return AudioStreamFormat::UNKNOWN;
}
}
AudioStreamFrameData fromProtoAudioData(const cmvr::api::AudioData& audio) {
AudioStreamFrameData frame;
frame.data.assign(audio.data().begin(), audio.data().end());
frame.sample_rate = audio.sample_rate() > 0 ? audio.sample_rate() : 44100;
frame.channels = audio.channels() > 0 ? audio.channels() : 2;
frame.format = fromProtoAudioFormat(audio.format());
frame.codec = audio.codec();
frame.pts = audio.pts();
frame.nb_samples = audio.nb_samples();
return frame;
}
}
gRPCSpeakerServiceImpl::gRPCSpeakerServiceImpl(): dmgr_(DeviceManager::getInstance()) {}
@ -81,6 +109,48 @@ grpc::Status gRPCSpeakerServiceImpl::PlayAudio(grpc::ServerContext* context,
}
}
grpc::Status gRPCSpeakerServiceImpl::StreamAudio(grpc::ServerContext* context,
grpc::ServerReader<api::StreamSpeakerAudioCommand_Request>* reader,
api::StreamSpeakerAudioCommand_Feedback* response) {
try {
api::StreamSpeakerAudioCommand_Request request;
std::shared_ptr<AbstractSpeaker> dev;
std::string dev_id;
while (reader->Read(&request)) {
if (!dev) {
dev_id = request.header().device_id();
CMVR_LOG(INFO) << "[gRPCSpeakerServiceImpl] (StreamAudio): id=" << dev_id;
dev = dmgr_.getDevice<AbstractSpeaker>(dev_id);
if (!dev) {
return failResponse(response, "Speaker device not found: " + dev_id);
}
if (!dev->start()) {
return failResponse(response, "Failed to start speaker: " + dev_id);
}
}
if (!dev->pushAudioFrame(fromProtoAudioData(request.audio()))) {
dev->stopStreaming();
return failResponse(response, "Failed to push speaker audio frame: " + dev_id);
}
}
if (dev) {
dev->stopStreaming();
}
response->mutable_header()->set_success(true);
setCurrentTimestamp(response->mutable_header()->mutable_timestamp());
return grpc::Status::OK;
}
catch (const std::exception& e) {
response->mutable_header()->set_success(false);
response->mutable_header()->set_error_message(e.what());
setCurrentTimestamp(response->mutable_header()->mutable_timestamp());
return grpc::Status::OK;
}
}
grpc::Status gRPCSpeakerServiceImpl::StopPlayback(grpc::ServerContext* context,
const api::StopSpeakerCommand_Request* request, api::StopSpeakerCommand_Feedback* response) {
try {

16
protos/cmvr/api/common.proto
View File

@ -29,6 +29,22 @@ message CommandHeader {
}
}
message AudioData {
enum AudioFormat {
PCM = 0;
MP3 = 1;
AAC = 2;
WAV = 3;
}
bytes data = 1;
int32 sample_rate = 2;
int32 channels = 3;
AudioFormat format = 4;
string codec = 5;
int64 pts = 6;
int32 nb_samples = 7;
}
message ConfigParam {
string param_name = 1;

17
protos/cmvr/api/microphone_command.proto
View File

@ -11,6 +11,7 @@ message MicState {
int32 volume = 4;
string error_message = 5;
}
message GetMicStateCommand {
message Request {
CommandHeader.Request header = 1;
@ -21,6 +22,7 @@ message GetMicStateCommand {
MicState state = 2;
}
}
message StartMicRecordingCommand {
message Request {
CommandHeader.Request header = 1;
@ -59,10 +61,20 @@ message ResumeMicRecordingCommand {
}
}
message StreamMicAudioCommand {
message Request {
CommandHeader.Request header = 1;
}
message Feedback {
CommandHeader.Feedback header = 1;
AudioData audio = 2;
}
}
message SetMicPhoneVolumeCommand {
message Request {
CommandHeader.Request header = 1;
int32 volume = 2; // 音量值（0 ~ 100）
int32 volume = 2;
}
message Feedback {
CommandHeader.Feedback header = 1;
@ -75,7 +87,6 @@ message GetMicPhoneVolumeCommand {
}
message Feedback {
CommandHeader.Feedback header = 1;
int32 volume = 2; // 当前音量值
int32 volume = 2;
}
}

5
protos/cmvr/api/microphone_service.proto
View File

@ -4,17 +4,14 @@ import "cmvr/api/microphone_command.proto";
package cmvr.api;
service MicPhoneService {
// 基本控制
rpc GetStatus(GetMicStateCommand.Request) returns (GetMicStateCommand.Feedback);
rpc StartRecord(StartMicRecordingCommand.Request) returns (StartMicRecordingCommand.Feedback);
rpc StopRecord(StopMicRecordingCommand.Request) returns (StopMicRecordingCommand.Feedback);
rpc PauseRecord(PauseMicRecordingCommand.Request) returns (PauseMicRecordingCommand.Feedback);
rpc ResumeRecord(ResumeMicRecordingCommand.Request) returns (ResumeMicRecordingCommand.Feedback);
rpc StreamAudio(StreamMicAudioCommand.Request) returns (stream StreamMicAudioCommand.Feedback);
// 音量控制
rpc SetVolume(SetMicPhoneVolumeCommand.Request) returns (SetMicPhoneVolumeCommand.Feedback);
rpc GetVolume(GetMicPhoneVolumeCommand.Request) returns (GetMicPhoneVolumeCommand.Feedback);
}

45
protos/cmvr/api/speaker_command.proto
View File

@ -4,33 +4,15 @@ import "cmvr/api/common.proto";
package cmvr.api;
// 音频数据格式（与麦克风类似，但用于输出）
message AudioData {
enum AudioFormat {
PCM = 0;
MP3 = 1;
AAC = 2;
WAV = 3;
}
bytes data = 1; // 音频二进制数据
int32 sample_rate = 2; // 采样率（Hz）
int32 channels = 3; // 声道数
AudioFormat format = 4; // 音频格式
string codec = 5; // 编码方式
}
// 扬声器状态
message SpeakerState {
bool is_initialized = 1; // 是否已初始化
bool is_running = 2; // 是否正在播放
bool is_initialized = 1;
bool is_running = 2;
bool is_decoding = 3;
bool is_paused = 5;
int32 volume = 6; // 当前音量（0 ~ 100）
string error_message = 7; // 错误信息
int32 volume = 6;
string error_message = 7;
}
// 各种扬声器命令
message GetSpeakerStateCommand {
message Request {
CommandHeader.Request header = 1;
@ -47,7 +29,19 @@ message PlayAudioCommand {
CommandHeader.Request header = 1;
string audio_path = 2;
}
message Feedback { CommandHeader.Feedback header = 1; }
message Feedback {
CommandHeader.Feedback header = 1;
}
}
message StreamSpeakerAudioCommand {
message Request {
CommandHeader.Request header = 1;
AudioData audio = 2;
}
message Feedback {
CommandHeader.Feedback header = 1;
}
}
message StopSpeakerCommand {
@ -80,7 +74,7 @@ message ResumeSpeakerCommand {
message SetSpeakerVolumeCommand {
message Request {
CommandHeader.Request header = 1;
int32 volume = 2; // 音量值（0 ~ 100）
int32 volume = 2;
}
message Feedback {
CommandHeader.Feedback header = 1;
@ -93,7 +87,6 @@ message GetSpeakerVolumeCommand {
}
message Feedback {
CommandHeader.Feedback header = 1;
int32 volume = 2; // 当前音量值
int32 volume = 2;
}
}

5
protos/cmvr/api/speaker_service.proto
View File

@ -4,17 +4,14 @@ import "cmvr/api/speaker_command.proto";
package cmvr.api;
service SpeakerService {
// 基本控制
rpc GetStatus(GetSpeakerStateCommand.Request) returns (GetSpeakerStateCommand.Feedback);
rpc PlayAudio(PlayAudioCommand.Request) returns (PlayAudioCommand.Feedback);
rpc StreamAudio(stream StreamSpeakerAudioCommand.Request) returns (StreamSpeakerAudioCommand.Feedback);
rpc StopPlayback(StopSpeakerCommand.Request) returns (StopSpeakerCommand.Feedback);
rpc PausePlayback(PauseSpeakerCommand.Request) returns (PauseSpeakerCommand.Feedback);
rpc ResumePlayback(ResumeSpeakerCommand.Request) returns (ResumeSpeakerCommand.Feedback);
// 音量控制
rpc SetVolume(SetSpeakerVolumeCommand.Request) returns (SetSpeakerVolumeCommand.Feedback);
rpc GetVolume(GetSpeakerVolumeCommand.Request) returns (GetSpeakerVolumeCommand.Feedback);
}