diff --git a/backend/cpp/audio-cpp/CMakeLists.txt b/backend/cpp/audio-cpp/CMakeLists.txt index 3cf4d3a31..269e7b9e8 100644 --- a/backend/cpp/audio-cpp/CMakeLists.txt +++ b/backend/cpp/audio-cpp/CMakeLists.txt @@ -90,6 +90,7 @@ add_executable(${TARGET} capability_routing.cpp family_gate.cpp loaded_model.cpp + audio_io.cpp audio_units.cpp transcript_assembly.cpp inference_lane.cpp diff --git a/backend/cpp/audio-cpp/audio_io.cpp b/backend/cpp/audio-cpp/audio_io.cpp new file mode 100644 index 000000000..275523b0b --- /dev/null +++ b/backend/cpp/audio-cpp/audio_io.cpp @@ -0,0 +1,72 @@ +#include "audio_io.h" + +#include "loaded_model.h" + +#include "engine/framework/audio/wav_reader.h" +#include "engine/framework/audio/wav_writer.h" + +#include +#include + +namespace audiocpp_backend { + +engine::runtime::AudioBuffer read_audio_file(const std::string &path) { + if (path.empty()) { + throw ConfigError("audio-cpp: no input audio path was supplied"); + } + std::error_code ec; + if (!std::filesystem::exists(std::filesystem::path(path), ec)) { + throw ConfigError("audio-cpp: input audio does not exist: " + path); + } + engine::runtime::AudioBuffer buffer; + try { + const engine::audio::WavData wav = + engine::audio::read_wav_f32(std::filesystem::path(path)); + buffer.sample_rate = wav.sample_rate; + // AudioBuffer's own default is 1, and a reader that reports 0 channels + // still gave us an interleaving of one. + buffer.channels = wav.channels > 0 ? wav.channels : 1; + buffer.samples = wav.samples; + } catch (const std::exception &err) { + throw ConfigError("audio-cpp: cannot read " + path + + " as WAV: " + err.what()); + } + if (buffer.sample_rate <= 0) { + throw ConfigError("audio-cpp: " + path + + " declares a non-positive sample rate; every " + "timestamp derived from it would be zero"); + } + return buffer; +} + +void write_audio_file(const std::string &path, + const engine::runtime::AudioBuffer &audio) { + if (path.empty()) { + throw ConfigError("audio-cpp: no output path was supplied"); + } + const std::filesystem::path destination(path); + if (destination.has_parent_path()) { + // Best effort: a failure here shows up as a write failure below, with a + // message naming the file the caller actually asked for. + std::error_code ec; + std::filesystem::create_directories(destination.parent_path(), ec); + } + try { + engine::audio::write_pcm16_wav(destination, audio.sample_rate, + audio.channels > 0 ? audio.channels : 1, + audio.samples); + } catch (const std::exception &err) { + throw ConfigError("audio-cpp: cannot write " + path + ": " + err.what()); + } +} + +engine::runtime::AudioBuffer buffer_from_mono(std::vector samples, + int sample_rate) { + engine::runtime::AudioBuffer buffer; + buffer.sample_rate = sample_rate; + buffer.channels = 1; + buffer.samples = std::move(samples); + return buffer; +} + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/audio_io.h b/backend/cpp/audio-cpp/audio_io.h new file mode 100644 index 000000000..10f59d61c --- /dev/null +++ b/backend/cpp/audio-cpp/audio_io.h @@ -0,0 +1,35 @@ +#pragma once + +// Thin wrappers over the framework's public audio IO. Engine-linked, so this +// unit is built and tested through the CMake target rather than by +// backend/cpp/run-unit-tests.sh. The pure part of the arithmetic these +// wrappers feed lives in audio_units, which is stdlib-only and does have a +// standalone test. + +#include "engine/framework/runtime/session.h" + +#include +#include + +namespace audiocpp_backend { + +// Reads a WAV file at its native sample rate and channel count. Throws +// ConfigError when the file is missing, is not readable as WAV, or declares a +// non-positive sample rate: all three are user-fixable input problems rather +// than backend faults. +// +// A declared sample rate of zero is refused rather than passed on, because +// every downstream conversion in audio_units answers 0 for a non-positive rate. +// Accepting it would turn a corrupt header into a response full of zero +// timestamps, which reads as a real answer. +engine::runtime::AudioBuffer read_audio_file(const std::string &path); + +// Writes 16-bit PCM WAV, creating parent directories. Throws ConfigError when +// the destination cannot be written. +void write_audio_file(const std::string &path, + const engine::runtime::AudioBuffer &audio); + +engine::runtime::AudioBuffer buffer_from_mono(std::vector samples, + int sample_rate); + +} // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/audio_units.cpp b/backend/cpp/audio-cpp/audio_units.cpp index 66c6013a4..7bc07ab23 100644 --- a/backend/cpp/audio-cpp/audio_units.cpp +++ b/backend/cpp/audio-cpp/audio_units.cpp @@ -6,6 +6,14 @@ namespace audiocpp_backend { +std::int64_t interleaved_frame_count(std::size_t sample_count, int channels) { + const std::size_t lanes = channels > 0 ? static_cast(channels) + : static_cast(1); + // Truncating division is deliberate: a trailing partial frame is not a + // position every channel reached, so counting it would overstate the length. + return static_cast(sample_count / lanes); +} + std::int64_t samples_to_nanoseconds(std::int64_t samples, int sample_rate) { if (sample_rate <= 0) { return 0; diff --git a/backend/cpp/audio-cpp/audio_units.h b/backend/cpp/audio-cpp/audio_units.h index 57581ebef..376d12d4c 100644 --- a/backend/cpp/audio-cpp/audio_units.h +++ b/backend/cpp/audio-cpp/audio_units.h @@ -8,12 +8,24 @@ // VADSegment start,end : float seconds // DiarizeSegment start,end : float seconds +#include #include #include #include namespace audiocpp_backend { +// Frames in an interleaved buffer of `sample_count` floats laid out across +// `channels` channels. A frame is one per-channel position, which is the unit +// every duration and every span boundary in this backend is expressed in, so a +// stereo buffer must not report twice its real length: feeding sample_count +// straight to samples_to_seconds makes a 3 second stereo clip come back as 6. +// +// A non-positive channel count is treated as mono, matching +// engine::runtime::AudioBuffer's own default of 1 and keeping a reader that +// reports 0 channels from dividing by zero. +std::int64_t interleaved_frame_count(std::size_t sample_count, int channels); + // Returns 0 when sample_rate is not positive rather than dividing by zero. // Uses integer arithmetic so 44.1 kHz does not lose precision. std::int64_t samples_to_nanoseconds(std::int64_t samples, int sample_rate); diff --git a/backend/cpp/audio-cpp/audio_units_test.cpp b/backend/cpp/audio-cpp/audio_units_test.cpp index 76f57d023..ec74eeee0 100644 --- a/backend/cpp/audio-cpp/audio_units_test.cpp +++ b/backend/cpp/audio-cpp/audio_units_test.cpp @@ -177,7 +177,38 @@ static void test_s16le_odd_length() { check(s16le_to_f32(std::string()).empty(), "empty input yields no samples"); } +static void test_interleaved_frame_count() { + // Mono is a pass-through, which is the only case the VAD path exercises. + check(interleaved_frame_count(16000, 1) == 16000, "mono frames equal samples"); + // The case that matters: a stereo buffer holds two floats per position, so a + // one second 16 kHz stereo clip is 32000 floats and still one second. Handing + // the raw float count to samples_to_seconds reports two seconds instead. + check(interleaved_frame_count(32000, 2) == 16000, + "stereo frames are half the samples"); + check(samples_to_seconds(interleaved_frame_count(32000, 2), 16000) == 1.0f, + "a one second stereo clip measures one second, not two"); + check(interleaved_frame_count(48000, 3) == 16000, + "three channels divide by three"); + // engine::runtime::AudioBuffer defaults channels to 1, but a reader is free + // to report 0, and dividing by that is undefined rather than merely wrong. + check(interleaved_frame_count(1000, 0) == 1000, + "zero channels is treated as mono"); + check(interleaved_frame_count(1000, -2) == 1000, + "a negative channel count is treated as mono"); + // A dangling partial frame is not a position every channel reached. + check(interleaved_frame_count(3, 2) == 1, + "a trailing partial frame is not counted"); + check(interleaved_frame_count(0, 2) == 0, "an empty buffer has no frames"); + // Past 2^32 floats, so a size_t narrowed to 32 bits on the way in, or a + // signed 32-bit intermediate, shows up here rather than in a multi-hour + // recording nobody tests with. + check(interleaved_frame_count(static_cast(9000000000ULL), 2) == + 4500000000LL, + "a buffer beyond 2^32 floats counts frames without truncating"); +} + int main() { + test_interleaved_frame_count(); test_nanoseconds(); test_seconds(); test_s16le_round_trip(); diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index eb9280217..715d53a34 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -5,12 +5,14 @@ // src/ or tests/ is used: those are application internals, they are where // upstream expects churn, and upstream is Apache-2.0 while LocalAI is MIT. // -// This commit adds LoadModel/Free/Status over one audio.cpp model. The audio -// RPCs themselves land in later commits. +// This commit adds LoadModel/Free/Status plus the VAD and Diarize RPCs. The +// remaining audio RPCs land in later commits. #include "backend.pb.h" #include "backend.grpc.pb.h" +#include "audio_io.h" +#include "audio_units.h" #include "capability_routing.h" #include "inference_lane.h" #include "loaded_model.h" @@ -24,12 +26,14 @@ #include #include #include +#include #include #include #include #include #include #include +#include #include #include #include @@ -72,6 +76,14 @@ std::shared_ptr snapshot() { return g_model; } +// LocalAI's VADRequest carries raw floats and no sample rate at all. Every +// LocalAI VAD backend treats them as 16 kHz mono (backend/go/silero-vad/vad.go +// and backend/go/sherpa-onnx/backend.go both hardcode 16000) and core/backend +// feeds them from a 16 kHz pipeline, so the same assumption is made here rather +// than left implicit. If the proto ever grows a sample rate field, this is the +// constant to delete. +constexpr int kVadSampleRate = 16000; + // Parses ModelOptions.MainGPU into a device index. // // Not std::atoi: it returns 0 for anything unparseable, so "gpu1" or a device @@ -210,6 +222,157 @@ public: result->set_success(true); return GStatus::OK; } + + GStatus VAD(ServerContext *, const backend::VADRequest *request, + backend::VADResponse *response) override { + try { + // snapshot(), and the result is held for the whole call. Free may + // arrive mid-request; it drops the global's reference only, so this + // one keeps the model alive until the handler returns. + const auto model = snapshot(); + if (model == nullptr) { + return GStatus(grpc::StatusCode::FAILED_PRECONDITION, + "audio-cpp: no model is loaded; call LoadModel first"); + } + + audiocpp_backend::RequestShape shape; + // Without this the `task:` model option is dead: routing is + // otherwise derived from the RPC alone. + shape.pinned_task = model->pinned_task(); + + engine::runtime::TaskRequest task; + task.audio_input = audiocpp_backend::buffer_from_mono( + std::vector(request->audio().begin(), request->audio().end()), + kVadSampleRate); + + // The lane is taken BEFORE session_for, not after. session_for + // reads and writes an unsynchronised session cache and prepare() + // mutates the session itself, so both belong inside the lane; see + // the note on session_for in loaded_model.h. Bound to a named local + // because LaneEntry is immovable and C++17 elides the return. + audiocpp_backend::LaneEntry lane = model->acquire(0); + const auto session = + model->session_for(audiocpp_backend::Rpc::Vad, shape); + const auto result = audiocpp_backend::run_offline(session, task); + + // VADSegment.start/end are float SECONDS, not sample indices. + for (const auto &segment : result.speech_segments) { + auto *out = response->add_segments(); + out->set_start(audiocpp_backend::samples_to_seconds( + segment.span.start_sample, kVadSampleRate)); + out->set_end(audiocpp_backend::samples_to_seconds( + segment.span.end_sample, kVadSampleRate)); + } + return GStatus::OK; + } catch (const std::exception &err) { + return to_status(err); + } + } + + GStatus Diarize(ServerContext *, const backend::DiarizeRequest *request, + backend::DiarizeResponse *response) override { + try { + const auto model = snapshot(); + if (model == nullptr) { + return GStatus(grpc::StatusCode::FAILED_PRECONDITION, + "audio-cpp: no model is loaded; call LoadModel first"); + } + + audiocpp_backend::RequestShape shape; + shape.pinned_task = model->pinned_task(); + + // Lane then route then read, in that order, and the read is last on + // purpose. A family that cannot diarize at all should say so, not + // complain about the input file first: on a VAD-only model, reading + // the audio would surface "cannot read /tmp/x.wav" and send the + // operator hunting a file problem instead of a model choice. The + // read costs a lane wait in exchange, which is the same wait every + // capability refusal already pays because session_for needs the lane. + audiocpp_backend::LaneEntry lane = model->acquire(0); + const auto session = + model->session_for(audiocpp_backend::Rpc::Diarize, shape); + + // DiarizeRequest.dst is the INPUT path, despite the field name: the + // HTTP layer materialises the upload to a temp file and passes the + // path here. Nothing is written back. + auto audio = audiocpp_backend::read_audio_file(request->dst()); + const int sample_rate = audio.sample_rate; + // Read off the buffer before it is moved into the request below. + // Frames, not floats: a stereo input holds two floats per position + // and would otherwise report twice its real duration. + const std::int64_t frames = audiocpp_backend::interleaved_frame_count( + audio.samples.size(), audio.channels); + + engine::runtime::TaskRequest task; + // Moved, not copied: a long recording runs to tens of megabytes and + // this is its only owner. + task.audio_input = std::move(audio); + // Forwarded verbatim, and deliberately not validated against what + // the family can do. audio.cpp's sortformer diarizer takes its + // speaker count from the model package (the bundled variant is + // 4-speaker) and reads none of these keys, so today they have no + // effect. That is the documented LocalAI contract, not a silent + // failure: core/backend/diarization.go says outright that backends + // ignore the hints they do not act on. Forwarding them still means a + // family that does read them gets them, with no change here. The + // knobs that DO reach sortformer (speaker_threshold, + // speaker_min_frames, speaker_pad_frames, session_len_sec) are + // session options and arrive through the model YAML's + // `session.:` entries at load time, not per request. + if (request->num_speakers() > 0) { + task.options["num_speakers"] = + std::to_string(request->num_speakers()); + } + if (request->min_speakers() > 0) { + task.options["min_speakers"] = + std::to_string(request->min_speakers()); + } + if (request->max_speakers() > 0) { + task.options["max_speakers"] = + std::to_string(request->max_speakers()); + } + + const auto result = audiocpp_backend::run_offline(session, task); + + // Emitted verbatim, in the order the model produced them. NOT + // sorted, merged or de-overlapped: sortformer binarizes each speaker + // independently, so a turn nested inside another speaker's turn is + // correct output for overlapped speech. LocalAI is overlap-tolerant + // downstream (core/backend/diarization.go passes segments through + // and the RTTM renderer handles overlap), so smoothing here would + // destroy real information. + std::set speakers; + int id = 0; + for (const auto &turn : result.speaker_turns) { + auto *out = response->add_segments(); + out->set_id(id++); + // Seconds, like VADSegment. Only TranscriptSegment/Word are ns. + out->set_start(audiocpp_backend::samples_to_seconds( + turn.span.start_sample, sample_rate)); + out->set_end(audiocpp_backend::samples_to_seconds( + turn.span.end_sample, sample_rate)); + out->set_speaker(turn.speaker_id); + // text stays EMPTY, including when include_text is set. + // audio.cpp's SpeakerTurn carries a span and a speaker label + // only, and TaskResult has no per-segment text anywhere, so + // there is nothing truthful to put here. + speakers.insert(turn.speaker_id); + } + response->set_num_speakers(static_cast(speakers.size())); + response->set_duration( + audiocpp_backend::samples_to_seconds(frames, sample_rate)); + + // Only set when the family bundles transcription, which no + // diarization-only family does; kept because a combined family + // would fill it in and the field is documented as optional. + if (result.text_output.has_value()) { + response->set_language(result.text_output->language); + } + return GStatus::OK; + } catch (const std::exception &err) { + return to_status(err); + } + } }; void RunServer(const std::string &addr) { diff --git a/backend/cpp/audio-cpp/loaded_model.cpp b/backend/cpp/audio-cpp/loaded_model.cpp index 971159409..292863f55 100644 --- a/backend/cpp/audio-cpp/loaded_model.cpp +++ b/backend/cpp/audio-cpp/loaded_model.cpp @@ -423,6 +423,15 @@ LaneEntry LoadedModel::acquire(int requested_timeout_ms) { requested_timeout_ms)); } +engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request) { + if (session.offline == nullptr) { + throw CapabilityError("audio-cpp: no offline session for this request"); + } + session.offline->prepare(engine::runtime::build_preparation_request(request)); + return session.offline->run(request); +} + std::unique_ptr LoadedModel::acquire_owned(int requested_timeout_ms) { return std::make_unique( lane_, diff --git a/backend/cpp/audio-cpp/loaded_model.h b/backend/cpp/audio-cpp/loaded_model.h index 1a2af5894..c2e308091 100644 --- a/backend/cpp/audio-cpp/loaded_model.h +++ b/backend/cpp/audio-cpp/loaded_model.h @@ -172,4 +172,20 @@ private: int wait_budget_ceiling_ms_ = 0; }; +// Prepares and runs an offline session. prepare() is called for every run +// rather than once per session, because SessionPreparationRequest is derived +// from the request itself (audio contract, text, voice condition) and not from +// the model: a second request with a different sample rate or length would +// otherwise run against the first request's contract. +// +// Call it only while holding the model's lane, for the same reason session_for +// requires it: the session is not reentrant, and prepare() mutates it. +// +// Throws CapabilityError when the session is not an offline one. That should be +// unreachable through session_for, which already refuses a non-offline session +// for an offline route, and is checked anyway because the alternative is a null +// dereference. +engine::runtime::TaskResult run_offline(const LoadedModel::Session &session, + const engine::runtime::TaskRequest &request); + } // namespace audiocpp_backend