utun2/tools/chatgui/tests/test_voice_tx.cpp

/* Реальные codec/AGC/автоматы/очередь uasync; наблюдаем сетевую границу вместо отправки пакетов. */
extern "C" {
#include "utun_instance.h"
#include "radio/radio_audio.h"
#include "call/call_audio.h"
#include "opus_codec.h"
#include "debug_config.h"
#include "u_async.h"
#include "silero_vad.h"
#include "audio_compressor.h"
#include "chat/chat_setting.h"
}
#include <vector>
#include <string>
#include <array>
#include <cstring>
#include <cmath>
#include <cstdio>
#include <cstdlib>
#include <algorithm>
#include <atomic>
#include <thread>
#include <chrono>
#define REQUIRE(x) do { if (!(x)) { fprintf(stderr, "voice-tx:%d %s\n", __LINE__, #x); abort(); } } while (0)
static std::vector<char> events;
static std::vector<std::vector<uint8_t>> packets;
static float probability;
static int vad_windows;
static int radio_compressor_enabled;
static bool observe_vad;
static std::vector<float> vad_input;
static std::atomic<bool> stall_vad{false}, vad_stalled{false};
extern "C" silero_vad_t* __wrap_silero_vad_create_default() { return reinterpret_cast<silero_vad_t*>(1); }
extern "C" void __wrap_silero_vad_destroy(silero_vad_t*) {}
extern "C" void __wrap_silero_vad_reset(silero_vad_t*) {}
extern "C" int __wrap_silero_vad_process(silero_vad_t*, const float* samples, float* result) {
    if (stall_vad) { vad_stalled = true; while (stall_vad) std::this_thread::sleep_for(std::chrono::milliseconds(1)); }
    if (observe_vad) vad_input.insert(vad_input.end(), samples, samples + SILERO_VAD_WINDOW_SAMPLES);
    ++vad_windows; *result = probability; return 0;
}
extern "C" int __wrap_radio_talk_begin(UTUN_INSTANCE*, uint64_t) { events.push_back('B'); return 0; }
extern "C" int __wrap_radio_talk_end(UTUN_INSTANCE*, uint64_t) { events.push_back('F'); return 0; }
extern "C" int __wrap_radio_talk_send(UTUN_INSTANCE*, uint64_t, const uint8_t* data, int len) {
    events.push_back('M'); packets.emplace_back(data, data + len); return 0;
}
extern "C" int __wrap_call_send_media(UTUN_INSTANCE*, uint64_t, const uint8_t* data, int len) {
    packets.emplace_back(data, data + len); return 0;
}
extern "C" int __real_chat_setting_get_int(UTUN_INSTANCE*, const char*, int);
extern "C" int __wrap_chat_setting_get_int(UTUN_INSTANCE* inst, const char* key, int fallback) {
    if (!strcmp(key, "radio_compressor_enabled")) return radio_compressor_enabled;
    if (!strcmp(key, "call_compressor_enabled")) return 0;
    if (!strcmp(key, "radio_vad_hangover_ms")) return 0;
    return __real_chat_setting_get_int(inst, key, fallback);
}
static void check_packet(opus_codec_encoder_t* encoder, const int16_t* pcm, size_t index,
                         struct audio_compressor* compressor = nullptr, int channels = 1) {
    std::array<int16_t, 1920> amplified;
    if (compressor) {
        REQUIRE(audio_compressor_process_frame(compressor, pcm, amplified.data(), 960 * channels) == 0);
        pcm = amplified.data();
    }
    uint8_t data[1024]; int size = opus_codec_encode(encoder, pcm, 960, data, sizeof(data));
    REQUIRE(size > 0 && index < packets.size());
    REQUIRE(packets[index].size() == (size_t)size && !memcmp(packets[index].data(), data, size));
}

static struct audio_compressor* reference_compressor(int channels, int max_gain) {
    auto* ac = audio_compressor_create(); REQUIRE(ac);
    audio_compressor_config_t cfg{};
    cfg.sample_rate = 48000; cfg.channels = channels; cfg.block_duration_ms = 20;
    cfg.lookback_ms = 200; cfg.max_gain_db = max_gain; cfg.rise_rate_per_sec = 2; cfg.target_level = 0.25f;
    REQUIRE(audio_compressor_configure(ac, &cfg) == 0);
    return ac;
}

/* Проверяем фактический вход Silero; разбиение callback не должно менять AGC или ресемплер. */
static void check_vad_pcm(const int16_t* pcm, int frames, int channels, int max_gain) {
    auto* ac = reference_compressor(channels, max_gain);
    std::array<int16_t, 1920> amplified;
    std::vector<float> expected;
    for (int offset = 0; offset + 960 <= frames; offset += 960) {
        REQUIRE(audio_compressor_process_frame(ac, pcm + offset * channels, amplified.data(), 960 * channels) == 0);
        for (int i = 0; i < 960; i += 3) {
            float sum = 0;
            for (int j = 0; j < 3; ++j) {
                const int16_t* sample = amplified.data() + (i + j) * channels;
                sum += channels == 2 ? ((float)sample[0] + sample[1]) * 0.5f : sample[0];
            }
            expected.push_back(sum / (3.0f * 32768.0f));
        }
    }
    expected.resize(expected.size() / 512 * 512);
    REQUIRE(vad_input == expected);
    REQUIRE(audio_compressor_gain_db(ac) <= max_gain + 0.001f);
    audio_compressor_destroy(ac);
}

static void test_vad_compressor(UTUN_INSTANCE* instance, int channels, int max_gain) {
    debug_set_category_level(DEBUG_CATEGORY_RADIO, DEBUG_LEVEL_DEBUG);
    REQUIRE(chat_setting_set(instance, "radio_compressor_max_gain_db", "6") == 0);
    REQUIRE(chat_setting_set(instance, "radio_vad_compressor_max_gain_db", std::to_string(max_gain).c_str()) == 0);
    REQUIRE(chat_setting_set(instance, "radio_vad_compressor_max_gain_db", "61") == -2);
    REQUIRE(chat_setting_set(instance, "radio_compressor_max_gain_db", "61") == -2);
    radio_compressor_enabled = 1; observe_vad = true; probability = 0;
    packets.clear(); events.clear(); vad_input.clear();
    REQUIRE(radio_audio_start(7, 0) == 0 && radio_audio_capture_start(7, channels, 1) == 0);
    const int idle = 55 * 960, live = 4 * 960;
    std::vector<int16_t> pcm((idle + live) * channels);
    for (size_t i = 0; i < pcm.size(); ++i) pcm[i] = (int16_t)(240 * sin(i * 0.07));
    auto feed_chunks = [&](const int16_t* samples, int frames) {
        const int chunks[] = {1, 7, 65, 997, 1536, 11, 333};
        for (int offset = 0, chunk = 0; offset < frames; ++chunk) {
            int count = std::min(chunks[chunk % 7], frames - offset);
            REQUIRE(radio_audio_feed_pcm(7, samples + offset * channels, count * channels) == 0);
            offset += count;
        }
    };
    feed_chunks(pcm.data(), idle);
    REQUIRE(!radio_audio_transmitting());
    radio_audio_talk_begin(7); feed_chunks(pcm.data() + idle * channels, live); radio_audio_talk_end(7);
    check_vad_pcm(pcm.data(), idle + live, channels, max_gain);
    REQUIRE(*std::max_element(vad_input.begin(), vad_input.end()) > 240.0f / 32768.0f * 1.5f);
    uasync_poll(instance->ua, 0);
    REQUIRE(packets.size() == 34 && events.front() == 'B' && events.back() == 'F');
    auto* encoder = opus_codec_encoder_create(48000, channels); REQUIRE(encoder);
    opus_codec_encoder_bitrate_set(encoder, 32000);
    auto* tx = reference_compressor(channels, 6);
    for (int i = 0; i < 34; ++i) check_packet(encoder, pcm.data() + (idle - 28800 + i * 960) * channels, i, tx, channels);
    audio_compressor_destroy(tx); opus_codec_encoder_destroy(encoder);
    // Разрыв и новый захват должны убрать и неполный блок 20мс, и историю усиления.
    feed_chunks(pcm.data(), 800);
    radio_audio_capture_discontinuity(7); vad_input.clear();
    feed_chunks(pcm.data(), 2897); check_vad_pcm(pcm.data(), 2897, channels, max_gain);
    auto reset_input = vad_input;
    radio_audio_capture_stop(); REQUIRE(radio_audio_capture_start(7, channels, 1) == 0); vad_input.clear();
    feed_chunks(pcm.data(), 2897); REQUIRE(vad_input == reset_input);
    radio_audio_stop(); uasync_poll(instance->ua, 0);
    observe_vad = false; radio_compressor_enabled = 0;
    REQUIRE(chat_setting_set(instance, "radio_compressor_max_gain_db", "15") == 0);
    REQUIRE(chat_setting_set(instance, "radio_vad_compressor_max_gain_db", "15") == 0);
    printf("voice TX: separate VAD gain=%ddB ch=%d, exact Silero input, independent TX/prefetch, chunking/reset PASS\n", max_gain, channels);
    debug_set_category_level(DEBUG_CATEGORY_RADIO, DEBUG_LEVEL_NONE);
}

/* Полные 600мс кольца, граница VAD внутри кадра и live должны совпасть с непрерывным Opus-потоком. */
static void test_prefetch(UTUN_INSTANCE* instance, int channels, int compressor_enabled = 0) {
    debug_set_category_level(DEBUG_CATEGORY_RADIO, DEBUG_LEVEL_INFO);
    packets.clear(); events.clear(); probability = 0; radio_compressor_enabled = compressor_enabled;
    REQUIRE(radio_audio_start(7, 0) == 0 && radio_audio_capture_start(7, channels, 1) == 0);
    const int idle = 33 * 1536, confirm = 2 * 1536, live = 959;
    std::vector<int16_t> pcm((idle + confirm + live) * channels);
    for (size_t i = 0; i < pcm.size(); ++i) pcm[i] = (int16_t)(10000 * sin(i * 0.07));
    REQUIRE(radio_audio_feed_pcm(7, pcm.data(), idle * channels) == 0);
    REQUIRE(!radio_audio_transmitting() && !instance->ua->posted_tasks_head);
    probability = 0.9f;
    int trigger = idle;
    for (; trigger < idle + confirm; ++trigger) {
        REQUIRE(radio_audio_feed_pcm(7, pcm.data() + trigger * channels, channels) == 0);
        if (radio_audio_transmitting()) break;
    }
    REQUIRE(radio_audio_transmitting());
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + (trigger + 1) * channels, live * channels) == 0);
    int pending = 0;
    for (auto* task = instance->ua->posted_tasks_head; task; task = task->next) ++pending;
    REQUIRE(pending == 3); // BEGIN + одна задача на 30 кадров prefetch + один live-кадр.
    radio_audio_capture_stop(); uasync_poll(instance->ua, 0);
    REQUIRE(packets.size() == 31 && events.size() == 33 && events.front() == 'B' && events.back() == 'F');
    auto* encoder = opus_codec_encoder_create(48000, channels); REQUIRE(encoder);
    opus_codec_encoder_bitrate_set(encoder, 32000);
    auto* tx = compressor_enabled ? reference_compressor(channels, 15) : nullptr;
    const int first = trigger - 28800;
    for (int i = 0; i < 31; ++i) check_packet(encoder, pcm.data() + (first + i * 960) * channels, i, tx, channels);
    audio_compressor_destroy(tx);
    opus_codec_encoder_destroy(encoder);
    radio_compressor_enabled = 0;

    // Разрыв очищает историю; короткий prefetch не дополняется тишиной на границе с live.
    packets.clear(); events.clear(); probability = 0;
    REQUIRE(radio_audio_capture_start(7, channels, 1) == 0);
    REQUIRE(radio_audio_feed_pcm(7, pcm.data(), 100 * channels) == 0);
    radio_audio_capture_discontinuity(7);
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + 100 * channels, 240 * channels) == 0);
    radio_audio_talk_begin(7);
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + 340 * channels, 120 * channels) == 0);
    radio_audio_talk_end(7); uasync_poll(instance->ua, 0);
    REQUIRE(events == std::vector<char>({'B','M','F'}));
    encoder = opus_codec_encoder_create(48000, channels); REQUIRE(encoder);
    opus_codec_encoder_bitrate_set(encoder, 32000);
    std::vector<int16_t> tail(960 * channels);
    memcpy(tail.data(), pcm.data() + 100 * channels, 360 * channels * sizeof(int16_t));
    check_packet(encoder, tail.data(), 0);
    opus_codec_encoder_destroy(encoder);

    // Следующая передача получает только новую паузу, а не уже переданный prefetch/хвост.
    packets.clear(); events.clear();
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + 460 * channels, 100 * channels) == 0);
    radio_audio_talk_begin(7);
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + 560 * channels, 100 * channels) == 0);
    radio_audio_talk_end(7); uasync_poll(instance->ua, 0);
    REQUIRE(events == std::vector<char>({'B','M','F'}));
    // Эталонный encoder сохраняет ту же историю предыдущего пакета.
    encoder = opus_codec_encoder_create(48000, channels); REQUIRE(encoder);
    opus_codec_encoder_bitrate_set(encoder, 32000);
    uint8_t unused[1024]; REQUIRE(opus_codec_encode(encoder, tail.data(), 960, unused, sizeof(unused)) > 0);
    std::fill(tail.begin(), tail.end(), 0);
    memcpy(tail.data(), pcm.data() + 460 * channels, 200 * channels * sizeof(int16_t));
    check_packet(encoder, tail.data(), 0);
    opus_codec_encoder_destroy(encoder);
    radio_audio_stop(); uasync_poll(instance->ua, 0);
    printf("voice TX: prefetch 600ms, exact ring/live order, short history, discontinuity, no replay ch=%d PASS\n", channels);
    debug_set_category_level(DEBUG_CATEGORY_RADIO, DEBUG_LEVEL_NONE);
}

static void test_prefetch_deferred(UTUN_INSTANCE* instance) {
    debug_set_category_level(DEBUG_CATEGORY_RADIO, DEBUG_LEVEL_INFO);
    packets.clear(); events.clear(); probability = 0;
    REQUIRE(radio_audio_start(7, 0) == 0 && radio_audio_capture_start(7, 1, 0) == 0);
    std::array<int16_t, 8 * 960> queued{};
    radio_audio_talk_begin(7); REQUIRE(radio_audio_feed_pcm(7, queued.data(), queued.size()) == 0);
    radio_audio_capture_stop(); REQUIRE(radio_audio_capture_start(7, 1, 1) == 0);
    const int idle = 33 * 1536, confirm = 2 * 1536;
    std::vector<int16_t> pcm(idle + confirm + 1);
    for (size_t i = 0; i < pcm.size(); ++i) pcm[i] = (int16_t)(9000 * sin(i * 0.09));
    REQUIRE(radio_audio_feed_pcm(7, pcm.data(), idle) == 0);
    probability = 0.9f;
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + idle, confirm) == 0);
    REQUIRE(!radio_audio_transmitting()); // Media-очередь занята: BEGIN отложен, история продолжает копиться.
    int pending = 0;
    for (auto* task = instance->ua->posted_tasks_head; task; task = task->next) ++pending;
    REQUIRE(pending == 10); // Только старые BEGIN + 8 media + FIN, новых задач пока нет.
    uasync_poll(instance->ua, 0); REQUIRE(packets.size() == 8);
    packets.clear(); events.clear();
    REQUIRE(radio_audio_feed_pcm(7, pcm.data() + idle + confirm, 1) == 0 && radio_audio_transmitting());
    radio_audio_capture_stop(); uasync_poll(instance->ua, 0);
    REQUIRE(packets.size() == 31 && events.front() == 'B' && events.back() == 'F');
    auto* encoder = opus_codec_encoder_create(48000, 1); REQUIRE(encoder);
    opus_codec_encoder_bitrate_set(encoder, 32000);
    const int first = idle + confirm - 28800;
    for (int i = 0; i < 30; ++i) check_packet(encoder, pcm.data() + first + i * 960, i);
    std::array<int16_t, 960> tail{}; tail[0] = pcm.back(); check_packet(encoder, tail.data(), 30);
    opus_codec_encoder_destroy(encoder);
    radio_audio_stop(); uasync_poll(instance->ua, 0);
    puts("voice TX: full media queue defers BEGIN and preserves complete prefetch PASS");
    debug_set_category_level(DEBUG_CATEGORY_RADIO, DEBUG_LEVEL_NONE);
}

int main() {
    debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN);
    UTUN_INSTANCE instance{}; instance.ua = uasync_create(); REQUIRE(instance.ua);
    REQUIRE(radio_audio_init(&instance) == 0 && call_audio_init(&instance) == 0);
    REQUIRE(radio_audio_start(7, 0) == 0 && radio_audio_capture_start(7, 1, 0) == 0);
    std::array<int16_t, 1920> input{};
    for (int i = 0; i < 1920; ++i) input[i] = (int16_t)(12000 * sin(i * 0.1));
    radio_audio_talk_begin(7); radio_audio_talk_begin(7);
    REQUIRE(radio_audio_feed_pcm(7, input.data(), 1200) == 0); radio_audio_talk_end(7);
    radio_audio_talk_begin(7); REQUIRE(radio_audio_feed_pcm(7, input.data() + 1200, 120) == 0); radio_audio_talk_end(7);
    uasync_poll(instance.ua, 0);
    REQUIRE(events == std::vector<char>({'B','M','M','F','B','M','F'}));
    auto* encoder = opus_codec_encoder_create(48000, 1); REQUIRE(encoder);
    check_packet(encoder, input.data(), 0);
    std::array<int16_t, 960> tail{}; memcpy(tail.data(), input.data() + 960, 240 * sizeof(int16_t));
    check_packet(encoder, tail.data(), 1);
    tail.fill(0); memcpy(tail.data(), input.data() + 1200, 120 * sizeof(int16_t)); check_packet(encoder, tail.data(), 2);
    // Разрыв не склеивает отсчёты до и после потери внутри одного Opus-кадра.
    packets.clear(); events.clear();
    radio_audio_talk_begin(7); radio_audio_feed_pcm(7, input.data(), 100);
    radio_audio_capture_discontinuity(7); radio_audio_feed_pcm(7, input.data() + 100, 100); radio_audio_talk_end(7);
    uasync_poll(instance.ua, 0);
    REQUIRE(events == std::vector<char>({'B','M','M','F'}));
    tail.fill(0); memcpy(tail.data(), input.data(), 100 * sizeof(int16_t)); check_packet(encoder, tail.data(), 0);
    tail.fill(0); memcpy(tail.data(), input.data() + 100, 100 * sizeof(int16_t)); check_packet(encoder, tail.data(), 1);
    opus_codec_encoder_destroy(encoder);
    // Capture teardown завершает принятую серию до FIN; следующее capture не получает старый хвост.
    packets.clear(); events.clear();
    radio_audio_talk_begin(7); radio_audio_feed_pcm(7, input.data(), 120); radio_audio_capture_stop();
    REQUIRE(radio_audio_capture_start(7, 1, 0) == 0);
    radio_audio_talk_begin(7); radio_audio_feed_pcm(7, input.data() + 120, 120); radio_audio_talk_end(7);
    uasync_poll(instance.ua, 0);
    REQUIRE(events == std::vector<char>({'B','M','F','B','M','F'}));
    // Старое прослушивание того же group ID отменено: старые BEGIN/media не попадают в новую сессию.
    packets.clear(); events.clear();
    radio_audio_talk_begin(7); radio_audio_feed_pcm(7, input.data(), 120); radio_audio_stop();
    REQUIRE(radio_audio_start(7, 1) == 0);
    uasync_poll(instance.ua, 0);
    REQUIRE(events.empty() && packets.empty());
    // У остановленного core ограничены не только media, но и быстрые PTT-команды.
    for (int i = 0; i < 1000; ++i) {
        radio_audio_talk_begin(7); radio_audio_feed_pcm(7, input.data(), 120); radio_audio_talk_end(7);
    }
    int pending = 0;
    for (auto* task = instance.ua->posted_tasks_head; task; task = task->next) ++pending;
    REQUIRE(pending <= 24); // 16 control + 8 media, FIN каждой принятой серии уже зарезервирован.
    uasync_poll(instance.ua, 0);
    packets.clear(); events.clear(); radio_audio_capture_stop();
    REQUIRE(radio_audio_capture_start(7, 1, 1) == 0 && radio_audio_vad_mode());
    auto feed = [&](int frames) {
        while (frames) { int count = std::min(frames, 1920); REQUIRE(radio_audio_feed_pcm(7, input.data(), count) == 0); frames -= count; }
    };
    probability = 0.9f; radio_audio_talk_begin(7); feed(3072); // Два окна VAD даже во время manual PTT.
    REQUIRE(vad_windows == 2); radio_audio_talk_end(7); REQUIRE(radio_audio_transmitting());
    probability = 0; feed(3072); REQUIRE(!radio_audio_transmitting()); // Первое окно начинает silence, второе завершает hangover.
    uasync_poll(instance.ua, 0);
    REQUIRE(std::count(events.begin(), events.end(), 'B') == 1 && std::count(events.begin(), events.end(), 'F') == 1);
    packets.clear(); events.clear();
    probability = 0.9f; feed(3072); REQUIRE(radio_audio_transmitting());
    radio_audio_talk_begin(7); probability = 0; feed(3072); REQUIRE(radio_audio_transmitting());
    radio_audio_talk_end(7); uasync_poll(instance.ua, 0);
    REQUIRE(std::count(events.begin(), events.end(), 'B') == 1 && std::count(events.begin(), events.end(), 'F') == 1);
    stall_vad = true;
    std::thread tx([&] { feed(1536); });
    for (int i = 0; i < 200 && !vad_stalled; ++i) std::this_thread::sleep_for(std::chrono::milliseconds(1));
    REQUIRE(vad_stalled);
    std::atomic<bool> rx_finished{false};
    std::thread rx([&] { int16_t out[1920]; REQUIRE(radio_audio_pull_pcm(7, out, 1920) == 1920); rx_finished = true; });
    for (int i = 0; i < 200 && !rx_finished; ++i) std::this_thread::sleep_for(std::chrono::milliseconds(1));
    REQUIRE(rx_finished); stall_vad = false; tx.join(); rx.join();
    radio_audio_stop(); uasync_poll(instance.ua, 0);
    test_prefetch(&instance, 1);
    test_prefetch(&instance, 2);
    test_prefetch(&instance, 1, 1);
    test_prefetch(&instance, 2, 1);
    test_prefetch_deferred(&instance);
    for (int channels : {1, 2}) for (int gain : {6, 30}) test_vad_compressor(&instance, channels, gain);
    // Старый FIN не обращается к новому ядру, даже когда allocator повторно использовал тот же адрес.
    REQUIRE(radio_audio_start(7, 1) == 0);
    radio_audio_talk_begin(7); uasync_poll(instance.ua, 0);
    radio_audio_talk_end(7); radio_audio_stop(); radio_audio_destroy(&instance);
    REQUIRE(radio_audio_init(&instance) == 0);
    events.clear(); uasync_poll(instance.ua, 0); REQUIRE(events.empty());
    // Prepared policy исключает повторный core AEC; mute применяется только к нужной части кадра.
    packets.clear(); REQUIRE(call_audio_start_prepared(42) == 0);
    REQUIRE(call_audio_start(42) == -1);
    REQUIRE(call_audio_feed_pcm(42, input.data(), 960) == -1);
    std::array<uint8_t, 960> mask{};
    std::fill(mask.begin() + 300, mask.end(), 1);
    REQUIRE(call_audio_feed_prepared_pcm(42, input.data(), mask.data()) == 0);
    uasync_poll(instance.ua, 0);
    encoder = opus_codec_encoder_create(48000, 1); REQUIRE(encoder);
    tail.fill(0); memcpy(tail.data(), input.data(), 300 * sizeof(int16_t)); check_packet(encoder, tail.data(), 0);
    call_audio_stop(); opus_codec_encoder_destroy(encoder);
    call_audio_destroy(&instance); radio_audio_destroy(&instance); uasync_destroy(instance.ua, 0);
    puts("voice TX: exact Opus tail, rapid bursts, discontinuity, capture policy, final mute PASS");
}